X-Git-Url: http://git.scottworley.com/pinch/blobdiff_plain/e8bd4979a02d7bb9e8fb7d978fd0e986d7ee60d1..530104d72eadd4a2700a5f8ba6d80dd7fed66d4e:/pinch.py?ds=inline diff --git a/pinch.py b/pinch.py index 1959f45..b5e3b81 100644 --- a/pinch.py +++ b/pinch.py @@ -11,6 +11,7 @@ import shlex import shutil import subprocess import sys +import tarfile import tempfile import types import urllib.parse @@ -18,6 +19,7 @@ import urllib.request import xml.dom.minidom from typing import ( + Callable, Dict, Iterable, List, @@ -96,6 +98,12 @@ class AliasPin(NamedTuple): pass +class SymlinkPin(NamedTuple): + @property + def release_name(self) -> str: + return 'link' + + class GitPin(NamedTuple): git_revision: str release_name: str @@ -108,7 +116,15 @@ class ChannelPin(NamedTuple): tarball_sha256: str -Pin = Union[AliasPin, GitPin, ChannelPin] +Pin = Union[AliasPin, SymlinkPin, GitPin, ChannelPin] + + +def copy_to_nix_store(v: Verification, filename: str) -> str: + v.status('Putting tarball in Nix store') + process = subprocess.run( + ['nix-store', '--add', filename], stdout=subprocess.PIPE) + v.result(process.returncode == 0) + return process.stdout.decode().strip() # type: ignore # (for old mypy) class AliasSearchPath(NamedTuple): @@ -119,6 +135,22 @@ class AliasSearchPath(NamedTuple): return AliasPin() +class SymlinkSearchPath(NamedTuple): + path: str + + # pylint: disable=no-self-use + def pin(self, _: Verification, __: Optional[Pin]) -> SymlinkPin: + return SymlinkPin() + + def fetch(self, v: Verification, _: Pin) -> str: + with tempfile.TemporaryDirectory() as td: + archive_filename = os.path.join(td, 'link.tar.gz') + os.symlink(self.path, os.path.join(td, 'link')) + with tarfile.open(archive_filename, mode='x:gz') as t: + t.add(os.path.join(td, 'link'), arcname='link') + return copy_to_nix_store(v, archive_filename) + + class GitSearchPath(NamedTuple): git_ref: str git_repo: str @@ -167,7 +199,10 @@ class ChannelSearchPath(NamedTuple): v, pin.tarball_url, Digest16(pin.tarball_sha256)) -SearchPath = Union[AliasSearchPath, GitSearchPath, ChannelSearchPath] +SearchPath = Union[AliasSearchPath, + SymlinkSearchPath, + GitSearchPath, + ChannelSearchPath] TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath] @@ -203,7 +238,7 @@ def fetch_channel( v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]: v.status('Fetching channel') request = urllib.request.urlopen(channel.channel_url, timeout=10) - channel_html = request.read() + channel_html = request.read().decode() forwarded_url = request.geturl() v.result(request.status == 200) # type: ignore # (for old mypy) v.check('Got forwarded', channel.channel_url != forwarded_url) @@ -512,11 +547,7 @@ def git_get_tarball( git.wait() v.result(git.returncode == 0 and xz.returncode == 0) - v.status('Putting tarball in Nix store') - process = subprocess.run( - ['nix-store', '--add', output_filename], stdout=subprocess.PIPE) - v.result(process.returncode == 0) - store_tarball = process.stdout.decode().strip() + store_tarball = copy_to_nix_store(v, output_filename) os.makedirs(os.path.dirname(cache_file), exist_ok=True) open(cache_file, 'w').write(store_tarball) @@ -590,31 +621,37 @@ K = TypeVar('K') V = TypeVar('V') -def filter_dict(d: Dict[K, V], fields: Set[K] - ) -> Tuple[Dict[K, V], Dict[K, V]]: +def partition_dict(pred: Callable[[K, V], bool], + d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]: selected: Dict[K, V] = {} remaining: Dict[K, V] = {} for k, v in d.items(): - if k in fields: + if pred(k, v): selected[k] = v else: remaining[k] = v return selected, remaining +def filter_dict(d: Dict[K, V], fields: Set[K] + ) -> Tuple[Dict[K, V], Dict[K, V]]: + return partition_dict(lambda k, v: k in fields, d) + + def read_config_section( conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]: mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = { 'alias': (AliasSearchPath, AliasPin), 'channel': (ChannelSearchPath, ChannelPin), 'git': (GitSearchPath, GitPin), + 'symlink': (SymlinkSearchPath, SymlinkPin), } SP, P = mapping[conf['type']] _, all_fields = filter_dict(dict(conf.items()), set(['type'])) pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields)) # Error suppression works around https://github.com/python/mypy/issues/9007 pin_present = pin_fields != {} or P._fields == () - pin = P(**pin_fields) if pin_present else None # type:ignore[call-arg] + pin = P(**pin_fields) if pin_present else None # type: ignore return SP(**remaining_fields), pin @@ -664,19 +701,24 @@ def pinCommand(args: argparse.Namespace) -> None: def updateCommand(args: argparse.Namespace) -> None: v = Verification() exprs: Dict[str, str] = {} - config = read_config_files(args.channels_file) - for section in config: - sp, pin = read_pinned_config_section(section, config[section]) - if isinstance(sp, AliasSearchPath): - continue + config = { + section: read_pinned_config_section(section, conf) for section, + conf in read_config_files( + args.channels_file).items()} + alias, nonalias = partition_dict( + lambda k, v: isinstance(v[0], AliasSearchPath), config) + + for section, (sp, pin) in nonalias.items(): + assert not isinstance(sp, AliasSearchPath) # mypy can't see through + assert not isinstance(pin, AliasPin) # partition_dict() tarball = sp.fetch(v, pin) exprs[section] = ( 'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' % - (config[section]['release_name'], tarball)) + (pin.release_name, tarball)) - for section in config: - if 'alias_of' in config[section]: - exprs[section] = exprs[str(config[section]['alias_of'])] + for section, (sp, pin) in alias.items(): + assert isinstance(sp, AliasSearchPath) # For mypy + exprs[section] = exprs[sp.alias_of] command = [ 'nix-env',