stratified_packager.toolbelt.zipping

Plugin-agnostic zip assembly, atomic publishing and archive-path validation helpers.

DO NOT import from the qgis package in this module (it must stay usable from worker threads and scripts/); standard library only.

Archive paths handled here are relative, slash-separated paths inside a zip (or below an output directory). Validation is strict and never sanitizes: callers that build paths from user expressions are expected to surface the reported violation instead of silently rewriting the path.

Module Attributes

PART_SUFFIX

Suffix of in-progress zip files; consumers only ever see fully published zips.

Functions

build_zip(zip_path, members, *, ...[, abort])

Assemble a zip from (source file, arcname) members.

case_insensitive_collisions(paths, /)

Group distinct paths that collide case-insensitively (the Windows filesystem rule).

filename_component_error(component, /)

Check one path component against the strict filename rules.

iter_file_members(root[, arc_prefix])

Yield (source file, arcname) pairs for every file below root, recursively.

publish_atomic(built, final, *[, abort])

Publish built at final via <final>.part and an atomic rename.

remove_stale_parts(directory, zip_names, /)

Remove leftover <name>.part files for this run's target zip names.

sha256_sidecar(path, /)

Write a <name>.sha256 checksum file next to path.

split_archive_path(path, /)

Validate a relative archive path and split it into its components.

stratified_packager.toolbelt.zipping._check_abort(abort, token)[source]

Raise when the abort callback signals cancellation.

Parameters:
  • abort (Callable[[], bool] | None) – Optional abort callback.

  • token (str) – Identifier carried in the exception (arcname or destination).

Raises:

OperationAbortedError – If abort is set and returned True.

Return type:

None

stratified_packager.toolbelt.zipping._copy_stream(src, dst, abort, token)[source]

Copy src into dst in chunks, polling abort between chunks.

Parameters:
  • src (IO[bytes]) – Readable binary stream.

  • dst (IO[bytes]) – Writable binary stream.

  • abort (Callable[[], bool] | None) – Optional abort callback polled between chunks.

  • token (str) – Identifier carried in the abort exception.

Raises:

OperationAbortedError – If abort returned True.

Return type:

None

stratified_packager.toolbelt.zipping._raise_path_error(path, reason)[source]

Raise the ValueError for an invalid archive path.

Parameters:
  • path (str) – The offending path, quoted in the message.

  • reason (str) – The violation description.

Return type:

tuple[str, ...]

Returns:

Never returns; the annotation only satisfies callers’ return idiom.

Raises:

ValueError – Always.

stratified_packager.toolbelt.zipping._write_members(archive, members, compression, compression_level, abort)[source]

Stream every member into the open archive.

Parameters:
  • archive (ZipFile) – Archive open for writing.

  • members (Iterable[tuple[Path, str]]) – (source file, arcname) pairs.

  • compression (int) – The archive-level compression constant.

  • compression_level (int) – The Deflate level (0 when stored).

  • abort (Callable[[], bool] | None) – Optional abort callback polled per member and per chunk.

Raises:

OperationAbortedError – If abort returned True.

Return type:

None

stratified_packager.toolbelt.zipping.build_zip(zip_path, members, *, compression_level, abort=None)[source]

Assemble a zip from (source file, arcname) members.

Compression is Deflate at compression_level, except level 0 which stores entries uncompressed (ZIP_STORED). Zip64 extensions are enabled, so archives and members beyond 4 GiB are supported. Sources are streamed in chunks, so large members do not load into memory.

Parameters:
  • zip_path (Path) – Destination zip file path (parent directory must exist; an existing file is overwritten).

  • members (Iterable[tuple[Path, str]]) – (source file, arcname) pairs; arcnames use / separators.

  • compression_level (int) – 0 to 9; 0 selects ZIP_STORED.

  • abort (Callable[[], bool] | None) – Optional callback polled before every member and between copy chunks; returning True aborts the build.

Return type:

Path

Returns:

zip_path.

Raises:

OperationAbortedError – If abort returned True; the partial file is removed before raising.

stratified_packager.toolbelt.zipping.case_insensitive_collisions(paths, /)[source]

Group distinct paths that collide case-insensitively (the Windows filesystem rule).

Parameters:

paths (Iterable[str]) – Slash-separated relative paths (duplicates are ignored).

Return type:

list[list[str]]

Returns:

One group per collision, each listing the distinct colliding spellings; empty when all paths are unique case-insensitively.

stratified_packager.toolbelt.zipping.filename_component_error(component, /)[source]

Check one path component against the strict filename rules.

The rules mirror utils.sanitize_filename() (illegal characters, reserved Windows device names, trailing dots/spaces, length) but validate instead of fixing — a violation is reported, never repaired.

Parameters:

component (str) – A single path component (no separators).

Return type:

str | None

Returns:

A human-readable violation description, or None when valid.

stratified_packager.toolbelt.zipping.iter_file_members(root, arc_prefix='', /)[source]

Yield (source file, arcname) pairs for every file below root, recursively.

Directory entries themselves are not yielded (zip directories materialize from member paths). Arcnames are slash-separated and prefixed with arc_prefix.

Parameters:
  • root (Path) – Directory to walk.

  • arc_prefix (str) – Prefix for every arcname (no trailing slash needed).

Yield:

(path, arcname) pairs, sorted for determinism.

stratified_packager.toolbelt.zipping.publish_atomic(built, final, *, abort=None)[source]

Publish built at final via <final>.part and an atomic rename.

The source is copied (chunked, abortable) to <final>.part in the destination directory, then renamed with os.replace() semantics — consumers of the destination directory only ever observe complete files. When built already lives on the destination filesystem the copy is still performed (simple and correct); callers building directly in the output directory pass a .part path as built instead and use pathlib.Path.replace() themselves.

Parameters:
  • built (Path) – The finished zip to publish.

  • final (Path) – The destination path (its parent is created if missing).

  • abort (Callable[[], bool] | None) – Optional callback polled between copy chunks; returning True aborts the publish.

Return type:

Path

Returns:

final.

Raises:

OperationAbortedError – If abort returned True; the .part file is removed before raising.

stratified_packager.toolbelt.zipping.remove_stale_parts(directory, zip_names, /)[source]

Remove leftover <name>.part files for this run’s target zip names.

Only the given names are touched — foreign .part files (other tools, other runs with different targets) are left alone.

Parameters:
  • directory (Path) – The output directory.

  • zip_names (Iterable[str]) – Final zip file names (e.g. north.zip) whose stale parts to drop.

Return type:

list[Path]

Returns:

The paths that were removed.

stratified_packager.toolbelt.zipping.sha256_sidecar(path, /)[source]

Write a <name>.sha256 checksum file next to path.

The content follows the sha256sum convention: <hex digest>  <filename> with a trailing newline, so standard tools can verify it. Written UTF-8: the filename may carry non-ASCII characters (e.g. an accented stratum name), which sha256sum reads back in the locale byte encoding.

Parameters:

path (Path) – The file to checksum (typically a published zip).

Return type:

Path

Returns:

The sidecar path.

stratified_packager.toolbelt.zipping.split_archive_path(path, /)[source]

Validate a relative archive path and split it into its components.

Accepts / and \ as separators. Rejects absolute paths (POSIX or Windows, including drive-letter and UNC forms), ./.. components, and any component violating filename_component_error().

Parameters:

path (str) – The relative path to validate (e.g. an evaluated gpkg or zip path).

Return type:

tuple[str, ...]

Returns:

The validated components, in order.

Raises:

ValueError – If the path is empty, absolute, escapes the root, or contains an invalid component; the message names the violation.

stratified_packager.toolbelt.zipping.PART_SUFFIX: Final = '.part'

Suffix of in-progress zip files; consumers only ever see fully published zips.

stratified_packager.toolbelt.zipping._COPY_CHUNK_SIZE: Final = 4194304

Chunk size (4 MiB) for abortable file copies.

stratified_packager.toolbelt.zipping._DRIVE_PREFIX: Final[re.Pattern[str]] = re.compile('^[A-Za-z]:')

Windows drive-letter prefix marking an absolute (or drive-relative) path.