Add symlink SearchPath type

[pinch] / pinch.py

diff --git a/pinch.py b/pinch.py
index 266daffb35045fffd8b490a6e17f24b97169b42c..37d37a7c137727bed59251a0b0b1619f19cbeab7 100644 (file)
--- a/pinch.py
+++ b/pinch.py
@@ -11,6 +11,7 @@ import shlex
 import shutil
 import subprocess
 import sys
 import shutil
 import subprocess
 import sys

+import tarfile

 import tempfile
 import types
 import urllib.parse
 import tempfile
 import types
 import urllib.parse


@@ -18,6 +19,7 @@ import urllib.request
 import xml.dom.minidom
 
 from typing import (
 import xml.dom.minidom
 
 from typing import (

+    Callable,

     Dict,
     Iterable,
     List,
     Dict,
     Iterable,
     List,


@@ -96,6 +98,12 @@ class AliasPin(NamedTuple):
     pass
 
 
     pass
 
 

+class SymlinkPin(NamedTuple):
+    @property
+    def release_name(self) -> str:
+        return 'link'
+
+

 class GitPin(NamedTuple):
     git_revision: str
     release_name: str
 class GitPin(NamedTuple):
     git_revision: str
     release_name: str


@@ -108,7 +116,15 @@ class ChannelPin(NamedTuple):
     tarball_sha256: str
 
 
     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()

 
 
 class AliasSearchPath(NamedTuple):
 
 
 class AliasSearchPath(NamedTuple):


@@ -119,6 +135,22 @@ class AliasSearchPath(NamedTuple):
         return AliasPin()
 
 
         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
 class GitSearchPath(NamedTuple):
     git_ref: str
     git_repo: str


@@ -167,7 +199,10 @@ class ChannelSearchPath(NamedTuple):
             v, pin.tarball_url, Digest16(pin.tarball_sha256))
 
 
             v, pin.tarball_url, Digest16(pin.tarball_sha256))
 
 

-SearchPath = Union[AliasSearchPath, GitSearchPath, ChannelSearchPath]
+SearchPath = Union[AliasSearchPath,
+                   SymlinkSearchPath,
+                   GitSearchPath,
+                   ChannelSearchPath]

 TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
 
 
 TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
 
 


@@ -512,11 +547,7 @@ def git_get_tarball(
             git.wait()
             v.result(git.returncode == 0 and xz.returncode == 0)
 
             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)
 
         os.makedirs(os.path.dirname(cache_file), exist_ok=True)
         open(cache_file, 'w').write(store_tarball)


@@ -590,24 +621,30 @@ K = TypeVar('K')
 V = TypeVar('V')
 
 
 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():
     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
 
 
             selected[k] = v
         else:
             remaining[k] = v
     return selected, remaining
 
 

-def read_search_path(
+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),
         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']))
     }
     SP, P = mapping[conf['type']]
     _, all_fields = filter_dict(dict(conf.items()), set(['type']))


@@ -618,6 +655,16 @@ def read_search_path(
     return SP(**remaining_fields), pin
 
 
     return SP(**remaining_fields), pin
 
 

+def read_pinned_config_section(
+        section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
+    sp, pin = read_config_section(conf)
+    if pin is None:
+        raise Exception(
+            'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
+            section)
+    return sp, pin
+
+

 def read_config(filename: str) -> configparser.ConfigParser:
     config = configparser.ConfigParser()
     config.read_file(open(filename), filename)
 def read_config(filename: str) -> configparser.ConfigParser:
     config = configparser.ConfigParser()
     config.read_file(open(filename), filename)


@@ -643,7 +690,7 @@ def pinCommand(args: argparse.Namespace) -> None:
         if args.channels and section not in args.channels:
             continue
 
         if args.channels and section not in args.channels:
             continue
 

-        sp, old_pin = read_search_path(config[section])
+        sp, old_pin = read_config_section(config[section])

 
         config[section].update(sp.pin(v, old_pin)._asdict())
 
 
         config[section].update(sp.pin(v, old_pin)._asdict())
 


@@ -654,23 +701,24 @@ def pinCommand(args: argparse.Namespace) -> None:
 def updateCommand(args: argparse.Namespace) -> None:
     v = Verification()
     exprs: Dict[str, str] = {}
 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_search_path(config[section])
-        if pin is None:
-            raise Exception(
-                'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
-                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"; }' %
         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',
 
     command = [
         'nix-env',