`http:` and `dav:`

Both run on the standard library, through a transport that can be replaced. The transport (HTTPTransport by default), HTTPResponse and the rules for redirects and credentials are those of pathapi.http; see its documentation.

`uripath.schemes.HTTPURIPath(*uris, **options)`

Bases: URIPath

http/https scheme: read/write access over HTTP (PUT/DELETE for writes/deletes, configurable via with_transport()), listing directories by scraping an Apache/nginx-style HTML index with a zero-dependency in-house parser (_DirectoryListingParser). Needs no third-party library: requests go through HTTPTransport unless a transport is given.

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

`backend` `instance-attribute`

`rmdir()`

Source code in src/uripath/schemes/_http/_path.py

def rmdir(self):
    # An empty directory's listing and a file whose body parses to zero entries look
    # alike from `_listdir()`: without this check rmdir() of a file would DELETE it
    # instead of raising NotADirectoryError.
    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/uripath/schemes/_http/_path.py

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

`unlink(missing_ok=False)`

Source code in src/uripath/schemes/_http/_path.py

def unlink(self, missing_ok=False):
    # A server that honours DELETE on a collection (WebDAV, RFC 4918) removes the whole
    # tree, and unlink() never removes a directory: refuse one. Over http: this uses the
    # HEAD `stat()` heuristic, which misses some collections (see `dav:` for reliable ones).
    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)`

This path with a backend that sends through a requests.Session (session=None makes a new one). headers and timeout in requests_args are the backend's; the others (auth, verify, proxies, cert, ...) go to the requests transport of pathapi-requests, which must be installed: without it this raises ImportError naming it. The caller closes the transport.

Source code in src/uripath/schemes/_http/_path.py

def with_session(
    self,
    session,
    write_method: str = "PUT",
    append_mode: str = "rewrite",
    **requests_args,
):
    """This path with a backend that sends through a `requests.Session`
    (`session=None` makes a new one). `headers` and `timeout` in
    `requests_args` are the backend's; the others (`auth`, `verify`,
    `proxies`, `cert`, ...) go to the `requests` transport of
    `pathapi-requests`, which must be installed: without it this raises
    `ImportError` naming it. The caller closes the transport."""
    from pathapi.http import get_transport

    headers = requests_args.pop("headers", None)
    timeout = requests_args.pop("timeout", _settings.DEFAULT_TIMEOUT)
    return self.with_transport(
        get_transport("requests")(session, **requests_args),
        headers=headers,
        timeout=timeout,
        write_method=write_method,
        append_mode=append_mode,
    )

`with_transport(transport, *, headers=None, timeout=_settings.DEFAULT_TIMEOUT, write_method='PUT', append_mode='rewrite')`

This path with a backend that sends through transport (any object with a send() method as pathapi.http.HTTPTransport.send(); the caller closes it). headers go with every request; write_method ("PUT", or e.g. "POST") uploads on close, and append_mode is "rewrite" or "patch".

Source code in src/uripath/schemes/_http/_path.py

def with_transport(
    self,
    transport,
    *,
    headers=None,
    timeout=_settings.DEFAULT_TIMEOUT,
    write_method: str = "PUT",
    append_mode: str = "rewrite",
):
    """This path with a backend that sends through `transport` (any object
    with a `send()` method as `pathapi.http.HTTPTransport.send()`;
    the caller closes
    it). `headers` go with every request; `write_method` (`"PUT"`, or
    e.g. `"POST"`) uploads on close, and `append_mode` is `"rewrite"` or
    `"patch"`."""
    backend = HTTPBackend(
        transport,
        headers=headers,
        timeout=timeout,
        write_method=write_method,
        append_mode=append_mode,
    )
    return type(self)(self, backend=backend)

`uripath.schemes.HTTPBackend(transport=None, *, headers=None, timeout=_settings.DEFAULT_TIMEOUT, write_method='PUT', append_mode='rewrite')`

Bases: HTTPClient

The transport and the settings shared by an HTTPURIPath tree (see with_transport() and with_session()): a pathapi.http.HTTPClient that takes a path or a URL for its address and adds write_method and append_mode.

transport is any object with a send() method as pathapi.http.HTTPTransport.send(); without one the backend builds the built-in HTTPTransport on first use and closes it with close() or at the end of a with block. A transport it was given stays the caller's. headers are sent with every request, request-specific headers merged over them by name; every request gets timeout=(10, 60) seconds unless timeout here or in the call says otherwise (None waits without limit). URL userinfo (http://user:password@host/) is never sent inside the request URL: it is stripped and sent as an Authorization: Basic header unless headers or the call name one.

The backend, through HTTPClient, follows redirects. GET, HEAD and OPTIONS follow up to 30; a redirect that changes the scheme, host or port drops Authorization (and nothing else). Any other method (PROPFIND, PUT, PATCH, DELETE, MKCOL, MOVE, a write_method such as POST) is sent without following: a 307 or 308 (for PROPFIND also a 301, 302 or 303) to the same scheme, host and port is sent once more with the same method and body, and any other 3xx, a redirect to another origin or a second redirect raises OSError(EIO).

Source code in src/uripath/schemes/_http/_backend.py

def __init__(
    self,
    transport=None,
    *,
    headers=None,
    timeout=_settings.DEFAULT_TIMEOUT,
    write_method: str = "PUT",
    append_mode: str = "rewrite",
):
    super().__init__(
        transport,
        headers=headers,
        timeout=timeout,
        user_agent=f"uripath/{__version__}",
    )
    self.write_method = write_method
    self.append_mode = append_mode

`slots = ('write_method', 'append_mode')` `class-attribute` `instance-attribute`

`append_mode = append_mode` `instance-attribute`

`write_method = write_method` `instance-attribute`

`repr()`

Source code in src/uripath/schemes/_http/_backend.py

def __repr__(self) -> str:
    return f"{type(self).__name__}(write_method={self.write_method!r})"

`redirect(method, uri, resp, *, headers=None, stream=False, timeout=_UNSET)`

Send method to where resp, the answer to a request for uri, redirects to, and follow what it redirects to in turn. The request carries what uri is sent with, except that Authorization is left off whenever it goes to another scheme, host or port than uri's. A redirect with no usable Location is OSError(EIO).

Source code in src/uripath/schemes/_http/_backend.py

def redirect(
    self,
    method,
    uri: "HTTPURIPath|str",
    resp: HTTPResponse,
    *,
    headers=None,
    stream: bool = False,
    timeout=_UNSET,
) -> HTTPResponse:
    """Send `method` to where `resp`, the answer to a request for `uri`,
    redirects to, and follow what it redirects to in turn. The request
    carries what `uri` is sent with, except that `Authorization` is left
    off whenever it goes to another scheme, host or port than `uri`'s.
    A redirect with no usable `Location` is `OSError(EIO)`."""
    url, auth = split_userinfo(uri if isinstance(uri, str) else uri.as_uri(False))
    return super().redirect(
        method,
        url,
        resp,
        headers=headers,
        stream=stream,
        timeout=self.timeout if timeout is _UNSET else timeout,
        auth=auth,
    )

`request(method, uri, *, data=None, headers=None, stream=False, allow_redirects=None, timeout=_UNSET)`

Send method to uri and return the final response. A data that is not bytes (a file, an iterable) cannot be sent twice, so it is not re-sent after a redirect or a dropped connection. allow_redirects=False returns the first answer of a GET, HEAD or OPTIONS as it is. ValueError for a URL or header with a control character, before anything is sent.

With stream=False the body is read before this returns, and a decoded body over pathapi.http.MAX_REPLY_BYTES is OSError(EFBIG); with stream=True the caller reads it, and closes the response.

Source code in src/uripath/schemes/_http/_backend.py

def request(
    self,
    method,
    uri: "HTTPURIPath|str",
    *,
    data=None,
    headers=None,
    stream: bool = False,
    allow_redirects=None,
    timeout=_UNSET,
) -> HTTPResponse:
    """Send `method` to `uri` and return the final response. A `data`
    that is not `bytes` (a file, an iterable) cannot be sent twice, so it
    is not re-sent after a redirect or a dropped connection.
    `allow_redirects=False` returns the first answer of a `GET`, `HEAD`
    or `OPTIONS` as it is. `ValueError` for a URL or header with a control
    character, before anything is sent.

    With `stream=False` the body is read before this returns, and a decoded
    body over `pathapi.http.MAX_REPLY_BYTES` is `OSError(EFBIG)`; with
    `stream=True` the caller reads it, and closes the response."""
    url, auth = split_userinfo(uri if isinstance(uri, str) else uri.as_uri(False))
    return super().request(
        method,
        url,
        data=data,
        headers=headers,
        stream=stream,
        allow_redirects=allow_redirects,
        timeout=self.timeout if timeout is _UNSET else timeout,
        auth=auth,
    )

`uripath.schemes.HTTPWriteStream(path, exclusive=False)`

Bases: _UploadStream

Source code in src/uripath/schemes/_http/_streams.py

def __init__(self, path: "HTTPURIPath", exclusive: bool = False):
    super().__init__(path)
    # `exclusive` (mode "x"): the upload carries `If-None-Match: *`, and a
    # 412 reply means the file appeared since the caller looked.
    self._exclusive = exclusive
    self._ready = True

`uripath.schemes.HTTPAppendStream(path)`

Bases: _UploadStream

Source code in src/uripath/schemes/_http/_streams.py

def __init__(self, path: "HTTPURIPath"):
    super().__init__(path)
    if path.backend.append_mode == "rewrite":
        try:
            existing = path.read_bytes()
        except FileNotFoundError:
            existing = b""
        self.write(existing)
    else:
        # "patch" mode: stat the resource to get its size as start offset.
        # Drop a listing-derived hint first: an offset taken from a hint
        # would overwrite the file wherever the listing was wrong.
        path._pop_stat_hint()
        try:
            stat, size_stated = path._head_stat()
        except FileNotFoundError:
            self._start_offset = 0
            self._existed = False
        else:
            if not size_stated:
                raise OSError(
                    _errno.EIO,
                    f"Cannot append to {path} in patch mode: the HEAD reply "
                    "has no Content-Length, so the end of the file is unknown",
                )
            self._start_offset = stat.st_size
            self._existed = True
    self._ready = True

`uripath.schemes.DAVURIPath(*uris, **options)`

Bases: HTTPURIPath

dav:/davs: scheme: WebDAV (RFC 4918) over HTTP(S). Extends HTTPURIPath with PROPFIND (stat/listdir -- real directory metadata, replacing HTTPURIPath'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 HTTPURIPath's HTTPBackend (transport + settings) -- 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/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