Add back create_archive/extract_archive

src/icepack/__init__.py

@@ -246,3 +246,82 @@ def _encrypt_path(self, src_path):
         except Exception:
             raise Exception('Failed to encrypt entry.')
         return tmp_path
+
+
+def create_archive(
+        src_path,
+        dst_path,
+        key_path,
+        compression=Compression.BZ2,
+        mode=False,
+        mtime=False,
+        recipients=None,
+        log=lambda msg: None):
+    """Convenience function for archive creation."""
+    src_path = src_path.resolve()
+    dst_path = dst_path.resolve()
+    base_path = src_path.parent
+    if src_path.is_file():
+        sources = [src_path]
+    elif src_path.is_dir():
+        sources = list(File.children(src_path))
+        sources.sort(key=_sort_key)
+    else:
+        raise Exception(f'Invalid source: {src_path}')
+    if recipients is not None:
+        public_key = (key_path / PUBLIC_KEY).read_text().strip()
+        signers = SSH.get_signers(key_path)
+        aliases = {alias: key for key, alias in signers if alias is not None}
+        resolved = list(map(lambda r: aliases.get(r, r), recipients))
+        resolved.insert(0, public_key)
+        recipients = resolved
+    if recipients is not None:
+        for r in recipients:
+            if not r.startswith('ssh-'):
+                raise Exception(f'Invalid recipient: {r}')
+    kwargs = {
+        'compression': compression,
+        'mode': mode,
+        'mtime': mtime,
+        'recipients': recipients,
+    }
+    with IcepackWriter(dst_path, key_path, **kwargs) as archive:
+        for source in sources:
+            log(source.relative_to(base_path))
+            archive.add_entry(source, base_path)
+        archive.add_metadata()
+
+
+def extract_archive(
+        src_path,
+        dst_path,
+        key_path,
+        mode=False,
+        mtime=False,
+        log=lambda msg: None):
+    """Convenience function for archive extraction."""
+    src_path = src_path.resolve()
+    dst_path = dst_path.resolve()
+    if not dst_path.is_dir():
+        raise Exception(f'Invalid destination: {dst_path}')
+    kwargs = {
+        'mode': mode,
+        'mtime': mtime,
+    }
+    with IcepackReader(src_path, key_path, **kwargs) as archive:
+        for entry in archive.metadata.entries:
+            log(entry.name)
+            archive.extract_entry(entry, dst_path)
+        if mtime:
+            # Fix directory mtimes
+            dirs = [e for e in archive.metadata.entries if e.is_dir()]
+            for d in dirs:
+                archive.extract_entry(d, dst_path)
+
+
+def _sort_key(path):
+    """Sort key for Paths."""
+    key = str(path).casefold()
+    if path.is_dir:
+        key += '/'
+    return key

src/icepack/cli.py

@@ -2,7 +2,7 @@
 
 import click
 
-from icepack import IcepackReader, IcepackWriter
+from icepack import IcepackReader, create_archive, extract_archive
 from icepack.helper import Age, File, SSH
 from icepack.meta import NAME, VERSION, SECRET_KEY, PUBLIC_KEY
 from icepack.model import Compression
@@ -67,38 +67,20 @@ def create(ctx, src, dst, compression, mode, mtime, recipient):
 
     SRC must be a file or directory, DST must be the archive file.
     """
-    src_path = Path(src).resolve()
-    dst_path = Path(dst).resolve()
-    if src_path.is_file():
-        sources = [src_path]
-    elif src_path.is_dir():
-        sources = list(File.children(src_path))
-        sources.sort(key=_sort_key)
-    else:
-        raise click.ClickException(f'Invalid source: {src_path}')
-    base = src_path.parent
+    src_path = Path(src)
+    dst_path = Path(dst)
     key_path = ctx.obj['config_path']
     _check_keys(key_path)
-    public_key = key_path / PUBLIC_KEY
-    signers = SSH.get_signers(key_path)
-    aliases = {alias: key for key, alias in signers if alias is not None}
-    recipients = [aliases[r] if r in aliases else r for r in recipient]
-    recipients.insert(0, public_key.read_text().strip())
-    for r in recipients:
-        if not r.startswith('ssh-'):
-            raise click.ClickException(f'Invalid recipient: {r}')
-    kwargs = {
-        'compression': compression,
-        'mode': mode,
-        'mtime': mtime,
-        'recipients': recipients,
-    }
     try:
-        with IcepackWriter(dst_path, key_path, **kwargs) as archive:
-            for source in sources:
-                click.echo(source.relative_to(base))
-                archive.add_entry(source, base)
-            archive.add_metadata()
+        create_archive(
+            src_path,
+            dst_path,
+            key_path,
+            compression=compression,
+            mode=mode,
+            mtime=mtime,
+            recipients=recipient,
+            log=click.echo)
     except Exception as e:
         raise click.ClickException(f'Archive creation failed: {e}')
 
@@ -120,26 +102,18 @@ def extract(ctx, src, dst, mode, mtime):
 
     SRC must be the archive file, DST must be a directory.
     """
-    src_path = Path(src).resolve()
-    dst_path = Path(dst).resolve()
-    if not dst_path.is_dir():
-        raise click.ClickException(f'Invalid destination: {dst_path}')
+    src_path = Path(src)
+    dst_path = Path(dst)
     key_path = ctx.obj['config_path']
     _check_keys(key_path)
-    kwargs = {
-        'mode': mode,
-        'mtime': mtime,
-    }
     try:
-        with IcepackReader(src_path, key_path, **kwargs) as archive:
-            for entry in archive.metadata.entries:
-                click.echo(entry.name)
-                archive.extract_entry(entry, dst_path)
-            if mtime:
-                # Fix directory mtimes
-                dirs = [e for e in archive.metadata.entries if e.is_dir()]
-                for d in dirs:
-                    archive.extract_entry(d, dst_path)
+        extract_archive(
+            src_path,
+            dst_path,
+            key_path,
+            mode=mode,
+            mtime=mtime,
+            log=click.echo)
     except Exception as e:
         raise click.ClickException(f'Archive extraction failed: {e}')
 
@@ -260,13 +234,5 @@ def _check_keys(key_path):
         raise click.ClickException(msg)
 
 
-def _sort_key(path):
-    """Sort key for Paths."""
-    key = str(path).casefold()
-    if path.is_dir:
-        key += '/'
-    return key
-
-
 if __name__ == '__main__':
     icepack()