Support old mypy version 0.701

[pinch] / pinch.py

diff --git a/pinch.py b/pinch.py
index 1959f450323f742420eb18c20c447c0815a3b3ee..b5e3b81567894101a9ee0d9cc9de34f253c15dd1 100644 (file)
--- 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',