111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
"""Dual-sink archiver that materializes a distribution tree.
|
|
|
|
The :class:`Archiver` mirrors the same logical entries into an on-disk
|
|
directory and/or a zip archive, and is usable as a context manager.
|
|
"""
|
|
|
|
import zipfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from types import TracebackType
|
|
from typing import Optional
|
|
|
|
|
|
class Archiver:
|
|
"""Dual sink that mirrors the same distribution tree into a directory and a zip archive.
|
|
|
|
On construction the archiver (optionally) prepares a directory on disk and
|
|
(optionally) opens a :class:`zipfile.ZipFile` for writing. Every subsequent
|
|
:meth:`push_file`/:meth:`push_text` call then writes the same logical entry
|
|
-- identified by its ``arcname``, a path relative to the distribution root
|
|
-- into both destinations, skipping whichever sink was not requested.
|
|
|
|
The archiver is usable as a context manager so that the underlying archive
|
|
is closed deterministically::
|
|
|
|
with Archiver(dist_dir, dist_zip) as arc:
|
|
arc.push_file(...)
|
|
"""
|
|
|
|
__dist_dir: Optional[Path]
|
|
__dist_zip: Optional[zipfile.ZipFile]
|
|
|
|
def __init__(self, dist_dir: Optional[Path], dist_zip: Optional[Path]) -> None:
|
|
"""Configure the sinks and acquire their backing resources.
|
|
|
|
:param dist_dir: Directory to mirror the tree into. Created (with
|
|
parents) when given; ``None`` disables the directory sink.
|
|
:param dist_zip: Path of the zip archive to create, truncated on open.
|
|
``None`` disables the zip sink.
|
|
"""
|
|
# make sure the distribution directory is existing
|
|
self.__dist_dir = dist_dir
|
|
if self.__dist_dir is not None:
|
|
self.__dist_dir.mkdir(parents=True, exist_ok=True)
|
|
# create zip instance
|
|
if dist_zip is not None:
|
|
self.__dist_zip = zipfile.ZipFile(dist_zip, "w", zipfile.ZIP_DEFLATED)
|
|
else:
|
|
self.__dist_zip = None
|
|
|
|
def push_dir(self, arcname: Path) -> None:
|
|
"""Create an empty directory entry in both sinks.
|
|
|
|
:param arcname: Directory path relative to the distribution root.
|
|
Parent directories are created as needed for the directory sink.
|
|
"""
|
|
if self.__dist_dir is not None:
|
|
target_filepath = self.__dist_dir / arcname
|
|
target_filepath.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if self.__dist_zip is not None:
|
|
self.__dist_zip.mkdir(arcname.as_posix())
|
|
|
|
def push_file(self, filepath: Path, arcname: Path) -> None:
|
|
"""Mirror an existing file into both sinks.
|
|
|
|
:param filepath: Source file on disk to copy / embed.
|
|
:param arcname: Destination path relative to the distribution root.
|
|
Missing parent directories are created for the directory sink.
|
|
"""
|
|
if self.__dist_dir is not None:
|
|
target_filepath = self.__dist_dir / arcname
|
|
target_filepath.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy(filepath, target_filepath)
|
|
|
|
if self.__dist_zip is not None:
|
|
self.__dist_zip.write(filepath, arcname.as_posix())
|
|
|
|
def push_text(self, text: str, arcname: Path) -> None:
|
|
"""Mirror in-memory text into both sinks.
|
|
|
|
:param text: UTF-8 text content to write.
|
|
:param arcname: Destination path relative to the distribution root.
|
|
Missing parent directories are created for the directory sink.
|
|
"""
|
|
if self.__dist_dir is not None:
|
|
target_filepath = self.__dist_dir / arcname
|
|
target_filepath.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(target_filepath, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
|
|
if self.__dist_zip is not None:
|
|
self.__dist_zip.writestr(arcname.as_posix(), text)
|
|
|
|
def __enter__(self) -> "Archiver":
|
|
"""Enter the context and return this archiver as the bound target."""
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: Optional[type[BaseException]],
|
|
exc_value: Optional[BaseException],
|
|
traceback: Optional[TracebackType],
|
|
) -> None:
|
|
"""Release the zip sink by closing its backing archive, if any.
|
|
|
|
Exceptions raised inside the ``with`` body are never suppressed.
|
|
"""
|
|
if self.__dist_zip is not None:
|
|
self.__dist_zip.close()
|