add cph list subcommand

news/236-list-subcommand

@@ -0,0 +1,19 @@
+### Enhancements
+
+* Add `cph list` to report artifact contents without prior extraction. (#236)
+
+### Bug fixes
+
+* <news item>
+
+### Deprecations
+
+* <news item>
+
+### Docs
+
+* <news item>
+
+### Other
+
+* <news item>

src/conda_package_handling/api.py

@@ -228,3 +228,13 @@ def get_pkg_details(in_file):
     else:
         raise ValueError(f"Don't know what to do with file {in_file}")
     return details
+
+
+def list_contents(in_file, verbose=False):
+    for format in SUPPORTED_EXTENSIONS.values():
+        if format.supported(in_file):
+            details = format.list_contents(in_file, verbose=verbose)
+            break
+    else:
+        raise ValueError(f"Don't know what to do with file {in_file}")
+    return details

src/conda_package_handling/cli.py

@@ -104,6 +104,14 @@ def build_parser():
         type=int,
         default=1,
     )
+    list_parser = sp.add_parser("list", help="list package contents", aliases=["l"])
+    list_parser.add_argument("archive_path", help="path to archive to inspect")
+    list_parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="Increase verbosity of output",
+    )
 
     return parser
 
@@ -135,6 +143,8 @@ def main(args=None):
             print("failed files:")
             pprint(failed_files)
             sys.exit(1)
+    elif args.subcommand in ("list", "l"):
+        api.list_contents(args.archive_path, verbose=args.verbose)
 
 
 if __name__ == "__main__":  # pragma: no cover

src/conda_package_handling/conda_fmt.py

@@ -16,7 +16,7 @@
 
 from . import utils
 from .interface import AbstractBaseFormat
-from .streaming import _extract
+from .streaming import _extract, _list
 
 CONDA_PACKAGE_FORMAT_VERSION = 2
 DEFAULT_COMPRESSION_TUPLE = (".tar.zst", "zstd", "zstd:compression-level=19")
@@ -142,3 +142,10 @@ def get_pkg_details(in_file):
         size = stat_result.st_size
         md5, sha256 = utils.checksums(in_file, ("md5", "sha256"))
         return {"size": size, "md5": md5, "sha256": sha256}
+
+    @staticmethod
+    def list_contents(fn, verbose=False, **kw):
+        components = utils.ensure_list(kw.get("components")) or ("info", "pkg")
+        if not os.path.isabs(fn):
+            fn = os.path.normpath(os.path.join(os.getcwd(), fn))
+        _list(fn, components=components, verbose=verbose)

src/conda_package_handling/interface.py

@@ -22,3 +22,8 @@ def create(prefix, file_list, out_fn, out_folder=os.getcwd(), **kw):  # pragma:
     @abc.abstractmethod
     def get_pkg_details(in_file):  # pragma: no cover
         raise NotImplementedError
+
+    @staticmethod
+    @abc.abstractmethod
+    def list_contents(in_file, verbose=False, **kw):  # pragma: no cover
+        raise NotImplementedError

src/conda_package_handling/streaming.py

@@ -4,7 +4,10 @@
 
 from __future__ import annotations
 
-from tarfile import TarError
+import io
+from contextlib import redirect_stdout
+from tarfile import TarError, TarFile, TarInfo
+from typing import Generator
 from zipfile import BadZipFile
 
 from conda_package_streaming.extract import exceptions as cps_exceptions
@@ -13,30 +16,54 @@
 from . import exceptions
 
 
-def _extract(filename: str, dest_dir: str, components: list[str]):
-    """
-    Extract .conda or .tar.bz2 package to dest_dir.
-
-    If it's a conda package, components may be ["pkg", "info"]
-
-    If it's a .tar.bz2 package, components must equal ["pkg"]
-
-    Internal. Skip directly to conda-package-streaming if you don't need
-    exception compatibility.
-    """
-
+def _stream_components(
+    filename: str,
+    components: list[str],
+    dest_dir: str = "",
+) -> Generator[tuple[TarFile, TarInfo], None, None]:
     if str(filename).endswith(".tar.bz2"):
         assert components == ["pkg"]
 
     try:
         with open(filename, "rb") as fileobj:
             for component in components:
                 # will parse zipfile twice
-                stream = package_streaming.stream_conda_component(
+                yield package_streaming.stream_conda_component(
                     filename, fileobj, component=component
                 )
-                extract_stream(stream, dest_dir)
     except cps_exceptions.CaseInsensitiveFileSystemError as e:
         raise exceptions.CaseInsensitiveFileSystemError(filename, dest_dir) from e
     except (OSError, TarError, BadZipFile) as e:
         raise exceptions.InvalidArchiveError(filename, f"failed with error: {str(e)}") from e
+
+
+def _extract(filename: str, dest_dir: str, components: list[str]):
+    """
+    Extract .conda or .tar.bz2 package to dest_dir.
+
+    If it's a conda package, components may be ["pkg", "info"]
+
+    If it's a .tar.bz2 package, components must equal ["pkg"]
+
+    Internal. Skip directly to conda-package-streaming if you don't need
+    exception compatibility.
+    """
+    for stream in _stream_components(filename, components, dest_dir=dest_dir):
+        try:
+            extract_stream(stream, dest_dir)
+        except cps_exceptions.CaseInsensitiveFileSystemError as e:
+            raise exceptions.CaseInsensitiveFileSystemError(filename, dest_dir) from e
+        except (OSError, TarError, BadZipFile) as e:
+            raise exceptions.InvalidArchiveError(filename, f"failed with error: {str(e)}") from e
+
+
+def _list(filename: str, components: list[str], verbose=True):
+    memfile = io.StringIO()
+    for component in _stream_components(filename, components):
+        for tar, _ in component:
+            with redirect_stdout(memfile):
+                tar.list(verbose=verbose)
+            component.close()
+    memfile.seek(0)
+    lines = sorted(memfile.readlines(), key=lambda line: line.split(None, 5)[-1])
+    print("".join(lines))

src/conda_package_handling/tarball.py

@@ -89,3 +89,9 @@ def get_pkg_details(in_file):
         size = stat_result.st_size
         md5, sha256 = utils.checksums(in_file, ("md5", "sha256"))
         return {"size": size, "md5": md5, "sha256": sha256}
+
+    @staticmethod
+    def list_contents(fn, verbose=False, **kw):
+        if not os.path.isabs(fn):
+            fn = os.path.normpath(os.path.join(os.getcwd(), fn))
+        streaming._list(str(fn), components=["pkg"], verbose=verbose)

tests/test_cli.py

@@ -1,3 +1,6 @@
+import io
+import os
+from contextlib import redirect_stdout
 from pathlib import Path
 
 import pytest
@@ -46,3 +49,16 @@ def test_import_main():
     """
     with pytest.raises(SystemExit):
         import conda_package_handling.__main__  # noqa
+
+
+@pytest.mark.parametrize(
+    "artifact,n_files",
+    [("mock-2.0.0-py37_1000.conda", 43), ("mock-2.0.0-py37_1000.tar.bz2", 43)],
+)
+def test_list(artifact, n_files):
+    "Integration test to ensure `cph list` works correctly."
+    memfile = io.StringIO()
+    with redirect_stdout(memfile):
+        cli.main(["list", os.path.join(data_dir, artifact)])
+    memfile.seek(0)
+    assert n_files == sum(1 for line in memfile.readlines() if line.strip())