Add --format option when publishing layer archives

First pass at #90 (still needs docs updates and additional tests)

src/venvstacks/cli.py

@@ -8,6 +8,7 @@
 import typer
 
 from .stacks import StackSpec, BuildEnvironment, _format_json, PackageIndexConfig
+from .pack_venv import ArchiveFormat, DEFAULT_ARCHIVE_FORMAT
 
 # Inspired by the Python 3.13+ `argparse` feature,
 # but reports `python -m venvstacks` whenever `__main__`
@@ -53,6 +54,7 @@ def handle_app_options() -> None:
 #   * _CLI_OPT_TRISTATE: boolean option, but allows None to indicate "not set"
 #   * _CLI_OPT_STR: optional string (defaulting to empty string to indicate "not set")
 #   * _CLI_OPT_STRLIST: multi-value list of strings
+#   * _CLI_OPT_ENUM: optional case-insensitive enum value
 #
 # The unit tests ensure the internal consistency of the CLI command annotations
 
@@ -121,6 +123,10 @@ def handle_app_options() -> None:
     bool,
     typer.Option(help="List archives that would be published")
 ]  # fmt: skip
+_CLI_OPT_ENUM_format = Annotated[
+    ArchiveFormat,
+    typer.Option(help="Archive compression format", case_sensitive=False)
+]  # fmt: skip
 
 # Selective processing of defined layers
 _CLI_OPT_STRLIST_include = Annotated[
@@ -222,11 +228,13 @@ def _publication_dry_run(
     build_env: BuildEnvironment,
     output_dir: str,
     tag_outputs: bool,
+    format: ArchiveFormat,
 ) -> None:
     base_output_path, dry_run_result = build_env.publish_artifacts(
         output_dir,
         dry_run=True,
         tag_outputs=tag_outputs,
+        format=format,
     )
     print("Archive creation skipped, reporting publishing request details:")
     print(_format_json(dry_run_result))
@@ -240,14 +248,18 @@ def _publish_artifacts(
     force: bool,
     dry_run: bool,
     tag_outputs: bool,
+    format: ArchiveFormat,
 ) -> None:
     if dry_run:
-        _publication_dry_run(build_env, output_dir, tag_outputs=tag_outputs)
+        _publication_dry_run(
+            build_env, output_dir, tag_outputs=tag_outputs, format=format
+        )
         return
     manifest_path, snippet_paths, archive_paths = build_env.publish_artifacts(
         output_dir,
         force=force,
         tag_outputs=tag_outputs,
+        format=format,
     )
     base_output_path = os.path.commonpath(
         [manifest_path, *snippet_paths, *archive_paths]
@@ -323,6 +335,8 @@ def build(
     local_wheels: _CLI_OPT_STRLIST_local_wheels = None,
     # Adjust naming of published archives and metadata files
     tag_outputs: _CLI_OPT_FLAG_tag_outputs = False,
+    # Adjust published archive format
+    format: _CLI_OPT_ENUM_format = DEFAULT_ARCHIVE_FORMAT,
     # Selective processing of defined layers
     include: _CLI_OPT_STRLIST_include = None,
     allow_missing: _CLI_OPT_FLAG_allow_missing = False,
@@ -372,7 +386,12 @@ def build(
         )
     build_env.create_environments(clean=clean, lock=lock)
     _publish_artifacts(
-        build_env, output_dir, dry_run=not publish, force=clean, tag_outputs=tag_outputs
+        build_env,
+        output_dir,
+        dry_run=not publish,
+        force=clean,
+        tag_outputs=tag_outputs,
+        format=format,
     )
 
 
@@ -448,6 +467,8 @@ def publish(
     dry_run: _CLI_OPT_FLAG_dry_run = False,
     # Adjust naming of published archives and metadata files
     tag_outputs: _CLI_OPT_FLAG_tag_outputs = False,
+    # Adjust published archive format
+    format: _CLI_OPT_ENUM_format = DEFAULT_ARCHIVE_FORMAT,
     # Selective processing of defined layers
     include: _CLI_OPT_STRLIST_include = None,
     allow_missing: _CLI_OPT_FLAG_allow_missing = False,
@@ -487,7 +508,12 @@ def publish(
             publish=True,
         )
     _publish_artifacts(
-        build_env, output_dir, force=force, dry_run=dry_run, tag_outputs=tag_outputs
+        build_env,
+        output_dir,
+        force=force,
+        dry_run=dry_run,
+        tag_outputs=tag_outputs,
+        format=format,
     )
 
 

src/venvstacks/pack_venv.py

@@ -107,10 +107,87 @@ def convert_symlinks(
     return relative_links, external_links
 
 
-def get_archive_path(archive_base_name: StrPath) -> Path:
-    """Report the name of the archive that will be created for the given base name."""
-    extension = ".zip" if _WINDOWS_BUILD else ".tar.xz"
-    return Path(os.fspath(archive_base_name) + extension)
+ProgressCallback = Callable[[str], None]
+
+
+class CompressionAlgorithm(StrEnum):
+    """Compression algorithm for published archive."""
+
+    UNCOMPRESSED = ""
+    BZIP2 = "bzip2"
+    GZIP = "gzip"
+    XZ = "xz"
+    ZIP = "zip"
+
+
+class ArchiveFormat(StrEnum):
+    """Archive publishing format."""
+
+    tar = "tar"
+    bz2 = "tar.bz2"
+    gz = "tar.gz"
+    xz = "tar.xz"
+    zip = "zip"
+
+    @property
+    def is_tar_format(self) -> bool:
+        """Whether this is a tar archive format."""
+        return self is not self.zip
+
+    @classmethod
+    def get_archive_format(cls, format: str | None) -> Self:
+        """Convert optional string value to a known archive format."""
+        if format is None:
+            return cls(DEFAULT_ARCHIVE_FORMAT)
+        return cls(format)
+
+    def get_archive_path(self, archive_base_name: StrPath) -> Path:
+        """Report the name of the archive that will be created for the given base name."""
+        return Path(os.fspath(archive_base_name) + f".{self}")
+
+    def make_archive(
+        self,
+        base_name: StrPath,
+        root_dir: StrPath,
+        base_dir: StrPath,
+        max_mtime: float | None = None,
+        progress_callback: ProgressCallback | None = None,
+    ) -> str:
+        """Create layer archive using this archive format and compression algorithm."""
+        if self.is_tar_format:
+            return _make_tar_archive(
+                base_name,
+                root_dir,
+                base_dir,
+                max_mtime,
+                progress_callback,
+                compress=str(self),
+            )
+        # Not a tar compression format -> emit a zipfile instead
+        return _make_zipfile(
+            base_name, root_dir, base_dir, max_mtime, progress_callback
+        )
+
+    def get_compression(self) -> str:
+        """Get compression algorithm for this archive format."""
+        return _ARCHIVE_COMPRESSION_ALGORITHMS[str(self)]
+
+
+_ARCHIVE_COMPRESSION_ALGORITHMS = {
+    str(ArchiveFormat.tar): str(CompressionAlgorithm.UNCOMPRESSED),
+    str(ArchiveFormat.bz2): str(CompressionAlgorithm.BZIP2),
+    str(ArchiveFormat.gz): str(CompressionAlgorithm.GZIP),
+    str(ArchiveFormat.xz): str(CompressionAlgorithm.XZ),
+    str(ArchiveFormat.zip): str(CompressionAlgorithm.ZIP),
+}
+
+
+if _WINDOWS_BUILD:
+    # No tar unpacking by default on windows, so use zipfile instead
+    DEFAULT_ARCHIVE_FORMAT = ArchiveFormat.zip
+else:
+    # Everywhere else, create XZ compressed tar archives
+    DEFAULT_ARCHIVE_FORMAT = ArchiveFormat.xz
 
 
 def _inject_postinstall_script(
@@ -187,68 +264,6 @@ def export_venv(
     return target_path
 
 
-if _WINDOWS_BUILD:
-    # No tar unpacking by default on windows, so use zipfile instead
-    _DEFAULT_ARCHIVE_FORMAT = "zip"
-else:
-    # Everywhere else, create XZ compressed tar archives
-    _DEFAULT_ARCHIVE_FORMAT = "xz"
-
-_COMPRESSION_FORMATS = {
-    "tar": "",
-    "tar.bz2": "bzip2",
-    "tar.gz": "gzip",
-    "tar.xz": "xz",
-}
-
-ProgressCallback = Callable[[str], None]
-
-
-class CompressionFormat(StrEnum):
-    """Compression format for published environment."""
-
-    UNCOMPRESSED = ""
-    BZIP2 = "bzip2"
-    GZIP = "gzip"
-    XZ = "xz"
-    ZIP = "zip"
-
-    @classmethod
-    def get_format(cls, format: str | None) -> Self:
-        """Get compression format for given value."""
-        if format is None:
-            return cls(_DEFAULT_ARCHIVE_FORMAT)
-        return cls(_COMPRESSION_FORMATS.get(format, format))
-
-    @property
-    def is_tar_format(self) -> bool:
-        """Whether this compression format is for a tar archive."""
-        return self is not self.ZIP
-
-    def make_archive(
-        self,
-        base_name: StrPath,
-        root_dir: StrPath,
-        base_dir: StrPath,
-        max_mtime: float | None = None,
-        progress_callback: ProgressCallback | None = None,
-    ) -> str:
-        """Create layer archive using this archive format."""
-        if self.is_tar_format:
-            return _make_tar_archive(
-                base_name,
-                root_dir,
-                base_dir,
-                max_mtime,
-                progress_callback,
-                compress=str(self),
-            )
-        # Not a tar compression format -> emit a zipfile instead
-        return _make_zipfile(
-            base_name, root_dir, base_dir, max_mtime, progress_callback
-        )
-
-
 def create_archive(
     source_dir: StrPath,
     archive_base_name: StrPath,
@@ -257,7 +272,7 @@ def create_archive(
     clamp_mtime: datetime | None = None,
     work_dir: StrPath | None = None,
     show_progress: bool = True,
-    format: CompressionFormat | None = None,
+    archive_format: ArchiveFormat | None = None,
 ) -> Path:
     """shutil.make_archive replacement, tailored for Python virtual environments.
 
@@ -305,17 +320,19 @@ def report_progress(_: Any) -> None:
             # To avoid filesystem time resolution quirks without relying on the resolution
             # details of the various archive formats, truncate mtime to exact seconds
             max_mtime = int(clamp_mtime.astimezone(timezone.utc).timestamp())
-        if format is None:
-            format = CompressionFormat.get_format(None)
-        archive_with_extension = format.make_archive(
+        if archive_format is None:
+            archive_format = DEFAULT_ARCHIVE_FORMAT
+        archive_with_extension = archive_format.make_archive(
             archive_path, env_path.parent, env_path.name, max_mtime, report_progress
         )
         if show_progress:
             # Ensure progress bar completion is reported, even if there's a discrepancy
             # between the number of paths found by `rglob` and the number of archive entries
             progress_bar.show(1.0)
     # The name query and the archive creation should always report the same archive name
-    assert archive_with_extension == os.fspath(get_archive_path(archive_base_name))
+    assert archive_with_extension == os.fspath(
+        archive_format.get_archive_path(archive_base_name)
+    )
     return Path(archive_with_extension)
 
 

src/venvstacks/stacks.py

@@ -582,6 +582,7 @@ class ArchiveBuildRequest:
     env_lock: EnvironmentLock
     env_path: Path
     archive_base_path: Path
+    archive_format: pack_venv.ArchiveFormat
     build_metadata: ArchiveBuildMetadata = field(repr=False)
     needs_build: bool = field(repr=False)
     # Previously built metadata when a new build is not needed
@@ -615,6 +616,7 @@ def define_build(
         source_path: Path,
         output_path: Path,
         target_platform: str,
+        archive_format: pack_venv.ArchiveFormat | None = None,
         tag_output: bool = False,
         previous_metadata: ArchiveMetadata | None = None,
         force: bool = False,
@@ -635,14 +637,16 @@ def define_build(
         # Work out the basic details of the build request (assuming no rebuild is needed)
         deployed_name = env_lock.get_deployed_name(env_name)
         build_iteration = last_build_iteration
+        if archive_format is None:
+            archive_format = pack_venv.DEFAULT_ARCHIVE_FORMAT
 
         def update_archive_name() -> tuple[Path, Path]:
             if tag_output:
                 base_name = f"{deployed_name}-{target_platform}-{build_iteration}"
             else:
                 base_name = deployed_name
             archive_base_path = output_path / base_name
-            built_archive_path = pack_venv.get_archive_path(archive_base_path)
+            built_archive_path = archive_format.get_archive_path(archive_base_path)
             return archive_base_path, built_archive_path
 
         archive_base_path, built_archive_path = update_archive_name()
@@ -677,6 +681,7 @@ def update_archive_name() -> tuple[Path, Path]:
             env_lock,
             env_path,
             archive_base_path,
+            archive_format,
             build_metadata,
             needs_build,
             archive_metadata,
@@ -1420,6 +1425,7 @@ def define_archive_build(
         tag_output: bool = False,
         previous_metadata: ArchiveMetadata | None = None,
         force: bool = False,
+        format: str | None = None,
     ) -> ArchiveBuildRequest:
         """Define an archive build request for this environment."""
         request = ArchiveBuildRequest.define_build(
@@ -1428,6 +1434,7 @@ def define_archive_build(
             self.build_path,
             output_path,
             target_platform,
+            pack_venv.ArchiveFormat.get_archive_format(format),
             tag_output,
             previous_metadata,
             force,
@@ -2320,6 +2327,7 @@ def publish_artifacts(
         force: bool = ...,
         tag_outputs: bool = ...,
         dry_run: Literal[False] = ...,
+        format: str | None = ...,
     ) -> PublishedArchivePaths: ...
     @overload
     def publish_artifacts(
@@ -2329,6 +2337,7 @@ def publish_artifacts(
         force: bool = ...,
         tag_outputs: bool = ...,
         dry_run: Literal[True] = ...,
+        format: str | None = ...,
     ) -> tuple[Path, StackPublishingRequest]: ...
     def publish_artifacts(
         self,
@@ -2337,6 +2346,7 @@ def publish_artifacts(
         force: bool = False,
         tag_outputs: bool = False,
         dry_run: bool = False,
+        format: str | None = None,
     ) -> PublishedArchivePaths | tuple[Path, StackPublishingRequest]:
         """Publish metadata and archives for specified layers."""
         layer_data: dict[
@@ -2368,6 +2378,7 @@ def publish_artifacts(
                         tag_output=tag_outputs,
                         previous_metadata=previous_metadata,
                         force=force and not dry_run,
+                        format=format,
                     ),
                 )
             )