http: and dav:

Both need the http extra.

pathlib_next.uri.schemes.http

DEFAULT_TIMEOUT = (10, 60) module-attribute

(connect, read) timeout, in seconds, HttpBackend sends with every request unless the caller supplies timeout (via with_session(..., timeout=...) / requests_args, or per request); timeout=None there restores requests' unbounded wait.

HttpBackend

Bases: NamedTuple

Per-instance requests.Session + extra request kwargs shared by an HttpPath tree (see with_session()).

Every request gets timeout=DEFAULT_TIMEOUT ((10, 60) seconds) unless requests_args or the call supplies timeout (None there waits forever). URL userinfo (http://user:password@host/) is never sent inside the request URL: it is stripped and sent as auth=(user, password) -- unless requests_args/the call pass their own auth or the session has session.auth set, which win as they did before.

append_mode = 'rewrite' class-attribute instance-attribute

requests_args instance-attribute

session instance-attribute

write_method = 'PUT' class-attribute instance-attribute

request(method, uri, **kwargs)

Source code in src/pathlib_next/uri/schemes/http.py

def request(self, method, uri: "HttpPath|str", **kwargs):
    url, auth = _split_userinfo(uri if isinstance(uri, str) else uri.as_uri(False))
    args = {**self.requests_args, **kwargs}
    if self.requests_args.get("headers") and kwargs.get("headers"):
        # Merged key by key: the caller's `with_session(headers=...)`
        # (an auth token, say) must survive a request that sends its
        # own headers (PROPFIND `Depth`, MOVE `Destination`); the
        # request's own values win on a shared key.
        args["headers"] = {**self.requests_args["headers"], **kwargs["headers"]}
    args.setdefault("timeout", DEFAULT_TIMEOUT)
    if (
        auth is not None
        and "auth" not in args
        and not getattr(self.session, "auth", None)
    ):
        args["auth"] = auth
    return self.session.request(method=method, url=url, **args)

HttpPath(*uris, **options)

Bases: UriPath

http/https scheme: read/write access over HTTP (PUT/DELETE for writes/deletes, configurable via with_session()), listing directories by scraping an Apache/nginx-style HTML index with a zero-dependency in-house parser (_DirectoryListingParser). Requires the http extra.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if 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(path).__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

backend instance-attribute

rmdir()

Source code in src/pathlib_next/uri/schemes/http.py

def rmdir(self):
    # An empty directory's listing and a file whose body happens to
    # parse to zero entries are indistinguishable from `_listdir()`
    # alone -- without this check, rmdir() on a *file* silently
    # DELETEd it instead of raising NotADirectoryError like
    # os.rmdir()/pathlib.Path.rmdir() do.
    if not self.is_dir():
        raise NotADirectoryError(self)
    # `_scandir()`, not the raw `_listdir()`: a listing's own "."/".."
    # rows are not children and must not make an empty directory
    # look non-empty.
    for _ in self._scandir():
        raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(self))
    self._delete()

stat(*, follow_symlinks=True, walk_up_last_modified=False)

Source code in src/pathlib_next/uri/schemes/http.py

def stat(self, *, follow_symlinks=True, walk_up_last_modified=False):
    hint = self._pop_stat_hint()
    if hint is not None:
        return hint

    with _translate_http_errors(self):
        # The caller's own spelling first: a trailing-slash path the
        # server answers is a directory, even when the slash-less URL
        # answers too (wsgidav serves both `/d` and `/d/` with 200).
        check = (
            [self, self.with_path(self.path.removesuffix("/"))]
            if self.path.endswith("/")
            else [self]
        )
        for uri in check:
            resp = self.backend.request(
                "HEAD", uri, allow_redirects=False, headers=_IDENTITY_ENCODING
            )
            resp.close()
            if resp.status_code == 405:
                # Some servers reject HEAD outright; fall back to GET.
                resp = self.backend.request(
                    "GET",
                    uri,
                    allow_redirects=False,
                    stream=True,
                    headers=_IDENTITY_ENCODING,
                )
                resp.close()
            if resp.status_code < 400:
                break

        if resp.is_redirect:
            resp = self.backend.request("HEAD", uri, headers=_IDENTITY_ENCODING)
            resp.close()
            if resp.status_code == 405:
                # Mirror the pre-redirect loop's HEAD-405 fallback --
                # without this, a server/proxy that rejects HEAD
                # everywhere (not just pre-redirect) surfaced
                # PermissionError for an existing, redirect-only path.
                resp = self.backend.request(
                    "GET", uri, stream=True, headers=_IDENTITY_ENCODING
                )
                resp.close()
        resp.raise_for_status()
        # From the final URL, once any redirect has been followed.
        is_dir = self._is_dir(resp)

    st_size = 0 if is_dir else int(resp.headers.get("Content-Length", 0))
    lm = resp.headers.get("Last-Modified")
    if lm is None and walk_up_last_modified:
        parent = self.parent
        if self != parent:
            try:
                entry = next(
                    filter(
                        lambda p: p.name.removesuffix("/") == self.name,
                        parent._listdir(),
                    )
                )
                if entry and entry.modified:
                    lm = entry.modified
            except (StopIteration, OSError):
                pass

    return FileStat(st_size=st_size, st_mtime=_utils.parsedate(lm), is_dir=is_dir)

unlink(missing_ok=False)

Source code in src/pathlib_next/uri/schemes/http.py

def unlink(self, missing_ok=False):
    # A server that honours DELETE on a collection (WebDAV, RFC 4918)
    # removes the whole tree, while pathlib's unlink() never removes a
    # directory -- refuse one. Limitation: over http: this relies on the
    # HEAD-based `stat()` heuristic, which cannot see every collection
    # (a wsgidav `/d/` answers HEAD with a plain 200 and reads as a
    # file); a child from `iterdir()` carries its listing's is_dir hint,
    # so that route is covered. Use `dav:` for reliable detection.
    if self.is_dir():
        raise IsADirectoryError(
            _errno.EISDIR, _os.strerror(_errno.EISDIR), str(self)
        )
    self._delete(missing_ok=missing_ok)

with_session(session, write_method='PUT', append_mode='rewrite', **requests_args)

Source code in src/pathlib_next/uri/schemes/http.py

def with_session(
    self,
    session: _req.Session,
    write_method: str = "PUT",
    append_mode: str = "rewrite",
    **requests_args,
):
    return type(self)(
        self, backend=HttpBackend(session, requests_args, write_method, append_mode)
    )

pathlib_next.uri.schemes.dav

DavPath(*uris, **options)

Bases: HttpPath

dav:/davs: scheme: WebDAV (RFC 4918) over HTTP(S). Extends HttpPath with PROPFIND (stat/listdir -- real directory metadata, replacing HttpPath's HTML-index scraping) and PUT/DELETE/MKCOL/MOVE (full write support). Requests go to the equivalent http:/https: URL (_wire_uri()); as_uri() still reports dav:/davs:. Reuses HttpPath's HttpBackend (session + requests_args) and the http extra -- no new dependency.

rmdir() enforces pathlib's "must be empty" contract with a depth-1 PROPFIND before DELETE (WebDAV DELETE is recursive by spec, RFC 4918, unlike pathlib.Path.rmdir()). The native recursive DELETE is still available -- and cheaper than the base class's client-side walk -- via rm(recursive=True), overridden below to issue a single request.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if 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(path).__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