Archives: zip:, tar:, archive:

archive+zip: and archive+tar: are second scheme names of ZipURIPath and TarURIPath (they pin the format, which archive: detects); ArchiveZipURIPath and ArchiveTarURIPath stay importable as aliases of those two classes.

uripath.schemes.ArchiveURIPath(*uris, **options)

Bases: DelegatingURIPath

Common base for zip:/tar:/archive: archive paths: <scheme>:<archive-uri>!/<inner-path> (Java-style separator; the <archive-uri> half is itself any absolute URI with an explicit scheme -- file:, http:, sftp:, ftp:, ... -- so archives are readable straight off any existing backend). segments/name/parent/glob/ ... all operate on the inner path; the outer archive handle is the backend, propagated through the normal backend machinery to every path derived from this one, and the members are read and written by a pathapi.archive.ArchivePath (ZipPath or TarPath once the format is known) on that handle.

Also registered directly as the archive: catch-all scheme (see __SCHEMES below): zip:/tar: (via ZipURIPath/TarURIPath, which just pin _core_cls) fix the format; plain archive: leaves it undecided and settles it on first use. Writes are gated on the handle being writable, which only a zip archive that is a local file: ever is -- so a tar-backed instance, whether reached via tar: or auto-detected via archive:, correctly raises NotImplementedError on any write attempt rather than silently misbehaving.

Source code in src/uripath/_uri.py

def __init__(self, *uris: URILike, **options):
    if options:
        self._check_options(options)
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | URI] = []
    for uri in uris:
        if uri is None or (isinstance(uri, (str, bytes)) and not uri):
            uri = ""
        if isinstance(uri, URI):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(uri).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

__SCHEMES = ('archive',) class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

backend property

__new__(*args, **kwargs)

Source code in src/uripath/schemes/_archive.py

def __new__(cls, *args, **kwargs):
    inst = super().__new__(cls, *args, **kwargs)
    if len(args) == 1 and isinstance(args[0], str) and not inst._raw_uris:
        # Keep the undecoded string for `_init` (see `_parse_archive_uri`).
        inst._raw_uris = [args[0]]
    return inst

as_uri(sanitize=False)

Source code in src/uripath/schemes/_archive.py

def as_uri(self, /, sanitize=False):
    if not self.source:
        # A derived relative path (`relative_to`): no archive to name.
        return super().as_uri(sanitize=sanitize)
    # Encoded so the string parses back to the same member: the inner
    # path's `%`, `?` and `#`, and a literal "!/" inside the outer URI.
    outer = self._outer_path()
    outer_uri = outer.as_uri(sanitize=sanitize)
    if not isinstance(outer, ArchiveURIPath):
        # A nested archive's outer keeps its own separator (it already
        # encodes any other "!/"); see `_split_archive_path`.
        outer_uri = outer_uri.replace(_SEP, "%21/")
    # The composer's own encoding: a name that is not UTF-8 (a tar member
    # written in another charset) keeps its bytes as `%XX`.
    inner = percent_encode(self.path, SAFE_PATH)
    # A "!/" inside a member name must never read as a separator once
    # this URI is itself the outer of a nested archive.
    inner = inner.replace(_SEP, "%21/")
    tail = self._format_parsed_parts(
        Source(None, None, None, None), "", self.query, self.fragment
    )
    return f"{self.source.scheme}:{outer_uri}{_SEP}{inner}{tail}"

batch()

A context manager that holds every change made to this archive while it is open and writes them in one replacement of the archive file when the outermost block ends without an exception; an exception leaves the archive as it was. Inside it every path of the archive reads the held changes, and the file is not touched until the block ends. A block inside another joins it. NotImplementedError when the archive cannot be written.

Source code in src/uripath/schemes/_archive.py

def batch(self) -> "_ty.ContextManager[None]":
    """A context manager that holds every change made to this archive
    while it is open and writes them in one replacement of the archive
    file when the outermost block ends without an exception; an exception
    leaves the archive as it was. Inside it every path of the archive
    reads the held changes, and the file is not touched until the block
    ends. A block inside another joins it. `NotImplementedError` when the
    archive cannot be written."""
    with self._delegated():
        return self._core_path.batch()

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/uripath/schemes/_archive.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."""
    with self._delegated():
        return self._core_path.readlink()

refresh()

Forgets what is held of the archive's outer 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/uripath/schemes/_archive.py

def refresh(self) -> None:
    """Forgets what is held of the archive's outer 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."""
    self._core_path.refresh()

rename(target)

Source code in src/uripath/schemes/_archive.py

def rename(self, target: "ArchiveURIPath | URI | str"):
    # A plain str target is a sibling rename (relative to self's parent), as in sftp
    # and ftp. An existing target is replaced as POSIX rename(2) does (a file over a
    # file, a directory over an empty one), never a duplicate member. Returns the new path.
    target = self._rename_target(target)
    core = self._core_path
    moved = core.with_segments(target.path.lstrip("/"))
    destination = self._from_parsed_parts(
        self.source, target.path.lstrip("/"), "", ""
    )
    with self._delegated((str(moved), str(destination))):
        renamed = core.rename(moved)
    return self._from_parsed_parts(
        self.source, renamed.as_posix().lstrip("/"), "", ""
    )

rm(recursive=False, missing_ok=False, ignore_error=False, *, follow_symlinks=False, follow_binds=False)

Source code in src/uripath/schemes/_archive.py

def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: "bool | _ty.Callable[[Exception, _ty.Self], bool]" = False,
    *,
    follow_symlinks=False,
    follow_binds=False,
):
    # An archive holds no bindings, and a link member is an entry like
    # any other for a removal that does not follow it.
    check_follow("follow_symlinks", follow_symlinks)
    check_follow("follow_binds", follow_binds)
    if not (
        recursive
        and self._rm_tree(
            as_error_handler(ignore_error), follows=follow_symlinks is not False
        )
    ):
        return super().rm(
            recursive=recursive,
            missing_ok=missing_ok,
            ignore_error=ignore_error,
            follow_symlinks=follow_symlinks,
            follow_binds=follow_binds,
        )

uripath.schemes.ZipURIPath(*uris, **options)

Bases: ArchiveURIPath

zip: scheme (archive+zip: is a second name for it). Read/write: write support (new entries, overwriting existing entries, unlink/rmdir/rename) works when the outer archive is a local file: URI; every other outer scheme is read-only (fetched fully into memory first). Every mutation replaces the archive atomically (temp file + 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 entry is appended, deflated, to a byte copy of the archive (nothing recompressed); overwriting/deleting/renaming an existing entry rewrites the whole archive, since the standard library's zip module has no in-place entry mutation, keeping every other entry's metadata and compression method. The members are read and written by a pathapi.archive.ZipPath.

Source code in src/uripath/_uri.py

def __init__(self, *uris: URILike, **options):
    if options:
        self._check_options(options)
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | URI] = []
    for uri in uris:
        if uri is None or (isinstance(uri, (str, bytes)) and not uri):
            uri = ""
        if isinstance(uri, URI):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(uri).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

__SCHEMES = ('zip', 'archive+zip') class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

uripath.schemes.TarURIPath(*uris, **options)

Bases: ArchiveURIPath

tar: scheme (also handles .tar.gz/.tar.bz2/.tar.xz via the standard library's auto-detected "r:*" mode). Read-only. Members stored with a ./ prefix are addressed without it. archive+tar: is the same scheme under a second name. The members are read by a pathapi.archive.TarPath.

Source code in src/uripath/_uri.py

def __init__(self, *uris: URILike, **options):
    if options:
        self._check_options(options)
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | URI] = []
    for uri in uris:
        if uri is None or (isinstance(uri, (str, bytes)) and not uri):
            uri = ""
        if isinstance(uri, URI):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(uri).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

__SCHEMES = ('tar', 'archive+tar') class-attribute instance-attribute

__slots__ = () class-attribute instance-attribute

uripath.schemes.MEMBER_SPOOL_BYTES is the size above which the content of an archive member read into memory is spooled to a temporary file; assign a different number to change it.