Start on 3.0.0

[pinch] / pinch.py

diff --git a/pinch.py b/pinch.py
index 9b845e3414df7a5c66a0cd940890a8562ebf7da8..ed33c928ae6a466c5e27a70ccf2a52d592bc8bc8 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


@@ -97,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


@@ -109,7 +116,24 @@ 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()  # type: ignore  # (for old mypy)
+
+
+def symlink_archive(v: Verification, path: str) -> str:
+    with tempfile.TemporaryDirectory() as td:
+        archive_filename = os.path.join(td, 'link.tar.gz')
+        os.symlink(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 AliasSearchPath(NamedTuple):
 
 
 class AliasSearchPath(NamedTuple):


@@ -120,6 +144,17 @@ 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:
+        return symlink_archive(v, self.path)
+
+

 class GitSearchPath(NamedTuple):
     git_ref: str
     git_repo: str
 class GitSearchPath(NamedTuple):
     git_ref: str
     git_repo: str


@@ -151,6 +186,8 @@ class ChannelSearchPath(NamedTuple):
 
         channel_html, forwarded_url = fetch_channel(v, self)
         table, new_gitpin = parse_channel(v, channel_html)
 
         channel_html, forwarded_url = fetch_channel(v, self)
         table, new_gitpin = parse_channel(v, channel_html)

+        if old_pin is not None and old_pin.git_revision == new_gitpin.git_revision:
+            return old_pin

         fetch_resources(v, new_gitpin, forwarded_url, table)
         ensure_git_rev_available(v, self, new_gitpin, old_revision)
         check_channel_contents(v, self, table, new_gitpin)
         fetch_resources(v, new_gitpin, forwarded_url, table)
         ensure_git_rev_available(v, self, new_gitpin, old_revision)
         check_channel_contents(v, self, table, new_gitpin)


@@ -168,7 +205,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]
 
 


@@ -204,7 +244,7 @@ def fetch_channel(
         v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
     v.status('Fetching channel')
     request = urllib.request.urlopen(channel.channel_url, timeout=10)
         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)
     forwarded_url = request.geturl()
     v.result(request.status == 200)  # type: ignore  # (for old mypy)
     v.check('Got forwarded', channel.channel_url != forwarded_url)


@@ -513,11 +553,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)


@@ -614,13 +650,14 @@ def read_config_section(
         'alias': (AliasSearchPath, AliasPin),
         'channel': (ChannelSearchPath, ChannelPin),
         'git': (GitSearchPath, GitPin),
         '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 == ()
     }
     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
 
 
     return SP(**remaining_fields), pin
 
 


@@ -692,8 +729,7 @@ def updateCommand(args: argparse.Namespace) -> None:
     command = [
         'nix-env',
         '--profile',
     command = [
         'nix-env',
         '--profile',

-        '/nix/var/nix/profiles/per-user/%s/channels' %
-        getpass.getuser(),
+        args.profile,

         '--show-trace',
         '--file',
         '<nix/unpack-channel.nix>',
         '--show-trace',
         '--file',
         '<nix/unpack-channel.nix>',


@@ -716,6 +752,8 @@ def main() -> None:
     parser_pin.set_defaults(func=pinCommand)
     parser_update = subparsers.add_parser('update')
     parser_update.add_argument('--dry-run', action='store_true')
     parser_pin.set_defaults(func=pinCommand)
     parser_update = subparsers.add_parser('update')
     parser_update.add_argument('--dry-run', action='store_true')

+    parser_update.add_argument('--profile', default=(
+        '/nix/var/nix/profiles/per-user/%s/channels' % getpass.getuser()))

     parser_update.add_argument('channels_file', type=str, nargs='+')
     parser_update.set_defaults(func=updateCommand)
     args = parser.parse_args()
     parser_update.add_argument('channels_file', type=str, nargs='+')
     parser_update.set_defaults(func=updateCommand)
     args = parser.parse_args()