Archive Paths API

pathapi.archive

Zip and tar archives as trees of paths, on the standard library alone (zipfile, tarfile).

ArchivePath, ZipPath and TarPath are Paths for the members of an archive file, which is any Path; they need no URI.

MAX_LINKS = 40 module-attribute

The most link members one lookup follows before it is a loop (OSError(ELOOP)).

ArchivePath(*segments, archive=None)

Bases: _WritableMembers, TreePath

A member path of an archive file, which is any Path (a LocalPath, a MemPath, a path on a remote store). The segments are the member path; the archive's format, zip or tar, is read from its extension or its first bytes on first use, and ZipPath and TarPath fix it. Building a path, printing it and joining names onto it read nothing.

A zip archive that is a local file can be written (new members, unlink/rmdir/rename, batch()); every other archive is read-only. One backend serves each archive file and format, so paths built separately for one file see each other's writes.

Source code in src/pathapi/archive/_path.py

def __init__(self, *segments: "str | ArchivePath", archive: object = None):
    text, donor = self._join(segments)
    if archive is not None:
        backend = self.backend_for(archive)
    elif donor is not None:
        backend = donor._backend
    else:
        raise TypeError("an archive path needs archive=")
    self._segments = self._parse(text)
    self._backend = backend

archive property

The archive file. Never decides the format.

backend property

The handle on the archive, shared by every path of it: settles the format of an ArchivePath on the first call, which reads the archive when its extension does not say.

as_uri()

Source code in src/pathapi/archive/_path.py

def as_uri(self) -> str:
    # Encoded so the string parses back to the same member: the member
    # path's `%`, `?` and `#`, and a literal "!/" inside the archive URI.
    outer = self.archive
    outer_uri = outer.as_uri()
    if not isinstance(outer, ArchivePath):
        # A nested archive's outer keeps its own separator (it already
        # encodes any other "!/").
        outer_uri = outer_uri.replace(_SEP, "%21/")
    member = _urlparse.quote(self.as_posix().lstrip("/"), safe="/")
    # A "!/" inside a member name must never read as a separator once
    # this URI is itself the outer of a nested archive.
    member = member.replace(_SEP, "%21/")
    return f"{self._SCHEME}:{outer_uri}{_SEP}{member}"

backend_for(archive, *, shared=True, identity=None) classmethod

The backend that serves archive, to pass as archive= of any path of the class: the backend of another path of the same archive is returned as it is. With shared the one backend of that archive file and format (found by its real path for a local file, else by identity or the node or URI of the path); with shared=False a backend of its own. Reads nothing: the format of ArchivePath is decided on first use.

Source code in src/pathapi/archive/_path.py

@classmethod
def backend_for(cls, archive: object, *, shared: bool = True, identity=None):
    """The backend that serves `archive`, to pass as `archive=` of any path
    of the class: the backend of another path of the same archive is
    returned as it is. With `shared` the one backend of that archive file
    and format (found by its real path for a local file, else by `identity`
    or the node or URI of the path); with `shared=False` a backend of its
    own. Reads nothing: the format of `ArchivePath` is decided on first
    use."""
    if isinstance(archive, (_ArchiveBackend, _UndecidedBackend)):
        return archive
    outer = _outer_file(archive)
    if cls._backend_cls is None:
        return _UndecidedBackend(outer, shared, identity)
    if not shared:
        return cls._backend_cls(outer)
    return _get_backend(cls._backend_cls, outer, identity)

iterdir()

Source code in src/pathapi/archive/_path.py

def iterdir(self):
    for name, _stat in self._scandir():
        yield self.with_segments(*self.segments, name)

readlink()

The text a link member stores, as a LinkTarget (OSError(EINVAL) when this member is not a link). Links on the way to it are followed.

Source code in src/pathapi/archive/_path.py

def readlink(self) -> LinkTarget:
    """The text a link member stores, as a `LinkTarget` (`OSError(EINVAL)`
    when this member is not a link). Links on the way to it are followed."""
    member = self._member
    if member is None:
        raise FileNotFoundError(str(self))
    snapshot = self._snapshot()
    member = self._resolve(snapshot, member, follow=False)
    if snapshot.stat(member) is None:
        raise FileNotFoundError(str(self))
    if member not in snapshot.links:
        raise OSError(_errno.EINVAL, "Invalid argument", str(self))
    text = snapshot.links[member]
    if text is None:
        raise OSError(_errno.EIO, "The link's text cannot be read", str(self))
    target = LinkTarget(text)
    target._text = text
    return target

refresh()

Forgets what is held of the archive file, so the next use reads it again. A non-local archive is read once and kept while any path to it lives, so this is how a later change at its source is seen; a local one is checked against the file on every use. Every path to the same archive shares the refresh.

Source code in src/pathapi/archive/_path.py

def refresh(self) -> None:
    """Forgets what is held of the archive file, so the next use reads it
    again. A non-local archive is read once and kept while any path to it
    lives, so this is how a later change at its source is seen; a local one
    is checked against the file on every use. Every path to the same
    archive shares the refresh."""
    backend = self._backend
    if isinstance(backend, _UndecidedBackend):
        backend = backend._resolved
        if backend is None:
            return  # nothing has been read yet
    backend.refresh()

stat(*, follow_symlinks=True)

Source code in src/pathapi/archive/_path.py

def stat(self, *, follow_symlinks=True):
    member = self._member
    if member is None:
        raise FileNotFoundError(str(self))
    # The root too: a missing or unreadable archive is not a directory
    # that exists, and a listing of it fails the same way.
    snapshot = self._snapshot()
    resolved = self._resolve(snapshot, member, follow_symlinks)
    if follow_symlinks and snapshot.links:
        self._check_enterable(snapshot, member, resolved)
    found = snapshot.stat(resolved)
    if found is None:
        raise FileNotFoundError(str(self))
    return found

LinkTarget

Bases: PurePosixPath

What readlink() of a link member returns: the stored text as a PurePosixPath, which spells it normalized (./a//b is a/b), and as path, which is the text exactly as stored. PathSyncer and copy() recreate a link from path.

path property

TarPath(*segments, archive=None)

Bases: ArchivePath

A member path of a tar archive, plain or compressed (tarfile's detected r:* mode), whatever its extension. Read-only. Members stored with a ./ prefix are addressed without it.

Source code in src/pathapi/archive/_path.py

def __init__(self, *segments: "str | ArchivePath", archive: object = None):
    text, donor = self._join(segments)
    if archive is not None:
        backend = self.backend_for(archive)
    elif donor is not None:
        backend = donor._backend
    else:
        raise TypeError("an archive path needs archive=")
    self._segments = self._parse(text)
    self._backend = backend

ZipPath(*segments, archive=None)

Bases: ArchivePath

A member path of a zip archive, whatever its extension. Read/write when the archive file is a local Path; every other archive is read-only (it is fetched whole into memory first). Every change replaces the archive atomically (a temporary file next to it, then os.replace), so the result is a new file: other hard links keep the old content, and an archive the caller may not write is refused first. A new member is appended, deflated, to a byte copy of the archive (nothing recompressed); overwriting, deleting or renaming a member rewrites the archive, keeping every other member's metadata and compression method.

Source code in src/pathapi/archive/_path.py

def __init__(self, *segments: "str | ArchivePath", archive: object = None):
    text, donor = self._join(segments)
    if archive is not None:
        backend = self.backend_for(archive)
    elif donor is not None:
        backend = donor._backend
    else:
        raise TypeError("an archive path needs archive=")
    self._segments = self._parse(text)
    self._backend = backend

detect_format(name, peek=None)

Detect "zip" vs "tar" from name's extension, falling back to magic-byte sniffing (PK header -> zip) via peek() -- called lazily, only when the extension is inconclusive, so callers backed by a remote Path don't pay for a round trip in the common case. Shared by unpack_archive and ArchivePath so both use one detection policy.

Source code in src/pathapi/archive/_pack.py

def detect_format(name: str, peek: "_ty.Callable[[], bytes] | None" = None) -> str:
    """Detect "zip" vs "tar" from `name`'s extension, falling back to
    magic-byte sniffing (`PK` header -> zip) via `peek()` -- called lazily,
    only when the extension is inconclusive, so callers backed by a remote
    Path don't pay for a round trip in the common case. Shared by
    `unpack_archive` and `ArchivePath` so both use one detection policy."""
    name_lower = name.lower()
    if name_lower.endswith((".zip", ".jar")):
        return "zip"
    if name_lower.endswith((".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tar.xz")):
        return "tar"
    magic = peek() if peek is not None else b""
    return "zip" if magic.startswith(b"PK") else "tar"

make_archive(src, format, target)

Create an archive file from src at target.

Supports format='zip' and format='tar'. Operations run stream-first to support any Path implementation, for src as well as target. The archive is built in a temporary buffer and written to target only once complete, so a failure leaves an existing target untouched. Zip members always carry zip64 headers, so a member over 2 GiB is stored instead of failing after the fact.

A directory is a member of its own, empty ones included. A member carries the modification time its source reports (a zip date from 1980, a tar time in whole seconds; none reported gives 1980-01-01 or the epoch) and, in a tar, the permission bits it reports (0o644 for a file and 0o755 for a directory when it reports none). A tar member's size is the number of bytes read from the source, not what stat() says: a backend may not know it, or count it differently.

Source code in src/pathapi/archive/_pack.py

def make_archive(src: Path, format: str, target: Path) -> None:
    """Create an archive file from `src` at `target`.

    Supports format='zip' and format='tar'.
    Operations run stream-first to support any Path implementation, for
    `src` as well as `target`. The archive is built in a temporary buffer
    and written to `target` only once complete, so a failure leaves an
    existing `target` untouched. Zip members always carry zip64 headers,
    so a member over 2 GiB is stored instead of failing after the fact.

    A directory is a member of its own, empty ones included. A member carries
    the modification time its source reports (a zip date from 1980, a tar time
    in whole seconds; none reported gives 1980-01-01 or the epoch) and, in a
    tar, the permission bits it reports (0o644 for a file and 0o755 for a
    directory when it reports none). A tar member's size is the number of
    bytes read from the source, not what `stat()` says: a backend may not know
    it, or count it differently.
    """
    if format not in ("zip", "tar"):
        raise ValueError(f"Unsupported format: {format}")

    if src.is_file():
        members = [(src, src.name, None, False)]
    elif src.is_dir():
        members = _archive_members(src)
    else:
        raise FileNotFoundError(f"No such file or directory: {str(src)!r}")

    with _Spool() as buffer:
        if format == "zip":
            with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
                for file_path, arcname, listed, is_dir in members:
                    stat = listed if is_dir else _content_stat(file_path, listed)
                    if is_dir:
                        info = zipfile.ZipInfo(f"{arcname}/", _zip_date(stat))
                        # drwxr-xr-x, with the MS-DOS directory flag.
                        info.external_attr = (0o40755 << 16) | 0x10
                        archive.writestr(info, b"")
                        continue
                    info = zipfile.ZipInfo(arcname, _zip_date(stat))
                    info.compress_type = zipfile.ZIP_DEFLATED
                    # force_zip64: the size is unknown when the entry opens,
                    # and without it zipfile raises once 2 GiB were written.
                    with archive.open(info, "w", force_zip64=True) as dest_f:
                        with file_path.open("rb") as src_f:
                            _copy_chunks(src_f, dest_f)
        else:
            with tarfile.open(fileobj=buffer, mode="w") as archive:
                for file_path, arcname, listed, is_dir in members:
                    if is_dir:
                        archive.addfile(_tar_info(arcname, listed, directory=True))
                        continue
                    info = _tar_info(
                        arcname, _content_stat(file_path, listed), directory=False
                    )
                    # The header comes first and holds the size, so the
                    # content is read once into a buffer to count it.
                    with _Spool() as content:
                        with file_path.open("rb") as src_f:
                            info.size = _copy_chunks(src_f, content)
                        content.seek(0)
                        archive.addfile(info, content)
        buffer.seek(0)
        with target.open("wb") as out_f:
            shutil.copyfileobj(buffer, out_f)

unpack_archive(archive, dest)

Extract archive file into dest directory.

Supports format detection from filename. Operations run stream-first to support any Path implementation. Members whose name would land outside dest (a .. part, or on a Windows-flavoured dest a drive such as D:x or C:..) are skipped. On a Windows-flavoured dest so is a member with a part Windows would store under another name (a trailing dot or space) or send to a device (NUL, CON, COM1, ...), with a UserWarning naming it. A non-seekable archive stream (e.g. an HTTP response) is buffered first. A tar hard link, or a symlink to a regular file inside the archive, is extracted as a regular file with the target member's content (no link is created, so dest needs no symlink support); a link that does not resolve to such a member (it names nothing, a directory, or itself through other links) is skipped with a UserWarning.

Source code in src/pathapi/archive/_pack.py

def unpack_archive(archive: Path, dest: Path) -> None:
    """Extract `archive` file into `dest` directory.

    Supports format detection from filename.
    Operations run stream-first to support any Path implementation.
    Members whose name would land outside `dest` (a `..` part, or on a
    Windows-flavoured `dest` a drive such as `D:x` or `C:..`) are skipped.
    On a Windows-flavoured `dest` so is a member with a part Windows would
    store under another name (a trailing dot or space) or send to a device
    (`NUL`, `CON`, `COM1`, ...), with a `UserWarning` naming it.
    A non-seekable archive stream (e.g. an HTTP response) is buffered first.
    A tar hard link, or a symlink to a regular file inside the archive, is
    extracted as a regular file with the target member's content (no link
    is created, so `dest` needs no symlink support); a link that does not
    resolve to such a member (it names nothing, a directory, or itself
    through other links) is skipped with a `UserWarning`.
    """
    if not dest.exists():
        dest.mkdir(parents=True, exist_ok=True)

    def _peek() -> bytes:
        try:
            with archive.open("rb") as f:
                return f.read(4)
        except Exception:
            return b"PK"  # can't sniff -- preserve the historical "assume zip" default

    is_zip = detect_format(archive.name, _peek) == "zip"
    windows = is_windows_flavoured(dest)

    def member_parts(filename: str) -> "list[str] | None":
        parts = _safe_member_parts(filename, windows=windows)
        if (
            parts is None
            and windows
            and _safe_member_parts(filename, windows=True, rewrites=False) is not None
        ):
            warnings.warn(
                f"unpack_archive: skipped member {filename!r}: Windows would "
                "store it under another name or send it to a device",
                UserWarning,
                stacklevel=3,
            )
        return parts

    with archive.open("rb") as raw_in, _Spool() as spool:
        in_f = raw_in
        seekable = getattr(raw_in, "seekable", None)
        if seekable is None or not seekable():
            # zipfile seeks to the central directory, and tarfile's "r:*"
            # probe seeks back: neither works over a network stream.
            shutil.copyfileobj(raw_in, spool)
            spool.seek(0)
            in_f = spool
        if is_zip:
            with zipfile.ZipFile(in_f) as zip_ref:
                for member in zip_ref.infolist():
                    filename = member.filename
                    parts = member_parts(filename)
                    if parts is None:
                        continue

                    target_path = dest
                    for part in parts:
                        target_path = target_path / part

                    if member.is_dir() or filename.endswith("/"):
                        target_path.mkdir(parents=True, exist_ok=True)
                    else:
                        target_path.parent.mkdir(parents=True, exist_ok=True)
                        with target_path.open("wb") as out_f:
                            with zip_ref.open(member) as member_f:
                                _copy_chunks(member_f, out_f)
        else:
            with tarfile.open(fileobj=in_f, mode="r") as tar_ref:
                members = tar_ref.getmembers()
                links = None
                for member in members:
                    filename = member.name
                    parts = member_parts(filename)
                    if parts is None:
                        continue

                    target_path = dest
                    for part in parts:
                        target_path = target_path / part

                    if member.isdir():
                        target_path.mkdir(parents=True, exist_ok=True)
                    elif member.isfile():
                        target_path.parent.mkdir(parents=True, exist_ok=True)
                        with target_path.open("wb") as out_f:
                            member_f = tar_ref.extractfile(member)
                            if member_f is not None:
                                _copy_chunks(member_f, out_f)
                    elif member.islnk() or member.issym():
                        # The member a link names inside the archive, never the filesystem, resolved one hop
                        # at a time: `extractfile()` follows a chain by recursion and a link that names
                        # itself never ends.
                        if links is None:
                            links = _TarLinks(members)
                        target_member = links.target(member)
                        member_f = (
                            None
                            if target_member is None
                            else tar_ref.extractfile(target_member)
                        )
                        if member_f is None:
                            warnings.warn(
                                f"unpack_archive: skipped link {filename!r} -> "
                                f"{member.linkname!r}: not a regular file in "
                                "the archive",
                                UserWarning,
                                stacklevel=2,
                            )
                            continue
                        target_path.parent.mkdir(parents=True, exist_ok=True)
                        with member_f, target_path.open("wb") as out_f:
                            _copy_chunks(member_f, out_f)