HTTP API

pathapi.http

HTTP requests on the standard library, with a replaceable transport.

HTTPClient sends a request through a transport (an object with send(): HTTPTransport on http.client, or one another distribution provides) and follows redirects under one rule for credentials: Authorization goes to the origin it was given for and to no other scheme, host or port. A reply comes back as an HTTPResponse. translate_errors() maps a failed request to the OSError types pathlib raises.

Importing this package loads http.client and ssl; import pathapi does not. Nothing here needs a third-party library.

DEFAULT_TIMEOUT = (10, 60) module-attribute

(connect, read) timeout, in seconds, a client sends with every request unless it is given timeout (HTTPClient(timeout=...) or per request); timeout=None waits without a limit.

TRANSPORT_ENV = 'PATHAPI_HTTP_TRANSPORT' module-attribute

Names the transport an HTTPClient builds for itself: builtin, or the name of one that an installed distribution declares in TRANSPORT_GROUP (requests, from pathapi-requests). Read when that transport is first needed; unset means builtin.

TRANSPORT_GROUP = 'pathapi.http_transports' module-attribute

The entry-point group of the HTTP transports. An entry's name is the transport's name (requests) and its value names a class that is built with no arguments.

HTTPClient(transport=None, *, headers=None, timeout=DEFAULT_TIMEOUT, user_agent=None)

A transport with the settings every request is sent with, and the rules for what happens after a redirect.

transport is any object with a send() method as HTTPTransport.send(); without one the client builds the transport that PATHAPI_HTTP_TRANSPORT names (the standard-library HTTPTransport by default) 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, over a User-Agent of user_agent (pathapi/<version> by default) and Accept: */*. Every request gets timeout=DEFAULT_TIMEOUT ((10, 60) seconds) unless timeout here or in the call says otherwise (None waits without limit).

The client, not the transport, 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 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/pathapi/http/_client.py

def __init__(
    self,
    transport: "_ty.Any" = None,
    *,
    headers: "_ty.Mapping[str, str] | None" = None,
    timeout: "float | tuple | None" = DEFAULT_TIMEOUT,
    user_agent: "str | None" = None,
):
    if transport is not None and not callable(getattr(transport, "send", None)):
        raise TypeError("a transport is an object with a send() method")
    self._transport = transport
    self._built = None
    self._transport_lock = _threading.Lock()
    self.headers = dict(headers or {})
    self.timeout = timeout
    self.user_agent = user_agent or f"pathapi/{__version__}"

headers = dict(headers or {}) instance-attribute

timeout = timeout instance-attribute

transport property

The transport: the one the client was given, or one built on first use. Read once per request.

user_agent = user_agent or f'pathapi/{__version__}' instance-attribute

close()

Closes the transport this client built for itself; one it was given stays the caller's. Harmless when called again.

Source code in src/pathapi/http/_client.py

def close(self) -> None:
    """Closes the transport this client built for itself; one it was given
    stays the caller's. Harmless when called again."""
    with self._transport_lock:
        built, self._built = self._built, None
        if built is not None:
            self._transport = None
    if built is not None:
        built.close()

redirect(method, url, resp, *, headers=None, stream=False, timeout=_UNSET, auth=None)

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

Source code in src/pathapi/http/_client.py

def redirect(
    self,
    method: str,
    url: str,
    resp: HTTPResponse,
    *,
    headers: "_ty.Mapping[str, str] | None" = None,
    stream: bool = False,
    timeout=_UNSET,
    auth: "tuple[str | bytes, str | bytes] | str | None" = None,
) -> HTTPResponse:
    """Send `method` to where `resp`, the answer to a request for `url`,
    redirects to, and follow what it redirects to in turn. The request
    carries what `url` is sent with, except that `Authorization` is left
    off whenever it goes to another scheme, host or port than `url`'s. A
    redirect with no usable `Location` is `OSError(EIO)`."""
    sent = self._sent_headers(headers, auth)
    location = _follow.redirect_location(url, resp)
    if location is None:
        raise _follow.refused_redirect(
            method, url, resp, None, "no usable Location"
        )
    return self._dispatch(
        method, location, sent, None, None, stream, timeout, first=url
    )

request(method, url, *, data=None, headers=None, stream=False, allow_redirects=None, timeout=_UNSET, auth=None)

Send method to url 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. auth is a (user, password) pair sent as Authorization: Basic, or a str sent as the Authorization value ("Bearer <token>"), to the origin of url unless a header names an Authorization. ValueError for a URL (credentials in it included) or a header with a control character, before anything is sent.

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

Source code in src/pathapi/http/_client.py

def request(
    self,
    method: str,
    url: str,
    *,
    data=None,
    headers: "_ty.Mapping[str, str] | None" = None,
    stream: bool = False,
    allow_redirects: "bool | None" = None,
    timeout=_UNSET,
    auth: "tuple[str | bytes, str | bytes] | str | None" = None,
) -> HTTPResponse:
    """Send `method` to `url` 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. `auth` is a `(user, password)` pair sent as
    `Authorization: Basic`, or a `str` sent as the `Authorization` value
    (`"Bearer <token>"`), to the origin of `url` unless a header names an
    `Authorization`. `ValueError` for a URL (credentials in it included)
    or a header with a control character, before anything is sent.

    With `stream=False` the body is read before this returns, and a decoded
    body over `MAX_REPLY_BYTES` is `OSError(EFBIG)`; with `stream=True` the
    caller reads it, and closes the response."""
    sent = self._sent_headers(headers, auth)
    return self._dispatch(method, url, sent, data, allow_redirects, stream, timeout)

HTTPResponse(status, reason, headers, url, body=b'', *, decode=True)

One response: status, reason, headers (a read-only mapping that ignores the case of a name), url (the URL the request went to) and the body, read with read() or iter_bytes(). A context manager that closes.

The body is decoded when the response says it is gzip or deflate encoded, unless the request set its own Accept-Encoding; headers are the headers as received, so Content-Length counts the encoded bytes. A body that ends before the length the server stated is OSError(EIO), never a shorter body that reads as complete.

A transport builds one with body as bytes, or as an object with read(amount) -> bytes (empty at the end) and close(), and decode=False for a body it already decoded or that was not asked for in a coding.

Source code in src/pathapi/http/_response.py

def __init__(
    self,
    status: int,
    reason: str,
    headers: "_ty.Mapping[str, str] | _ty.Iterable[tuple[str, str]]",
    url: str,
    body: "bytes | _ty.Any" = b"",
    *,
    decode: bool = True,
):
    self._closed = False
    self._ended = False
    self.status = int(status)
    self.reason = reason or ""
    self.headers = headers if isinstance(headers, Headers) else Headers(headers)
    self.url = url
    self._kept = False
    if isinstance(body, (bytes, bytearray, memoryview)):
        body = _Memory(body)
        self._kept = True  # in memory: readable after close
    self._source = body
    self._inflater = _inflater_for(self.headers) if decode else None
    self._leftover = b""

headers = headers if isinstance(headers, Headers) else Headers(headers) instance-attribute

reason = reason or '' instance-attribute

status = int(status) instance-attribute

url = url instance-attribute

buffered()

Reads the rest of the body into memory, closes the connection and returns this response, which can then be read after it is closed. A transport returns its response this way when send() is called with stream=False.

Source code in src/pathapi/http/_response.py

def buffered(self) -> "HTTPResponse":
    """Reads the rest of the body into memory, closes the connection and
    returns this response, which can then be read after it is closed. A
    transport returns its response this way when `send()` is called with
    `stream=False`."""
    return self._buffered()

close()

Releases the connection. Harmless when called again. A body that was not read to its end is abandoned, and its connection is closed.

Source code in src/pathapi/http/_response.py

def close(self) -> None:
    """Releases the connection. Harmless when called again. A body that was
    not read to its end is abandoned, and its connection is closed."""
    if not getattr(self, "_closed", True):
        self._closed = True
        self._source.close()

iter_bytes(chunk_size=CHUNK)

The rest of the body in pieces of at most chunk_size bytes, each as soon as it has arrived.

Source code in src/pathapi/http/_response.py

def iter_bytes(self, chunk_size: int = CHUNK) -> "_ty.Iterator[bytes]":
    """The rest of the body in pieces of at most `chunk_size` bytes, each
    as soon as it has arrived."""
    self._check_open()
    while True:
        chunk = self._leftover or self._next(chunk_size)
        if not chunk:
            return
        self._leftover = chunk[chunk_size:]
        yield chunk[:chunk_size]

read(size=-1)

Up to size bytes of the decoded body, fewer only at its end; the whole rest of it when size is negative or None.

Source code in src/pathapi/http/_response.py

def read(self, size: "int | None" = -1) -> bytes:
    """Up to `size` bytes of the decoded body, fewer only at its end; the
    whole rest of it when `size` is negative or `None`."""
    self._check_open()
    parts = []
    have = 0
    while size is None or size < 0 or have < size:
        wanted = CHUNK if size is None or size < 0 else size - have
        chunk = self._leftover or self._next(wanted)
        if not chunk:
            break
        if size is not None and 0 <= size < have + len(chunk):
            self._leftover = chunk[size - have :]
            chunk = chunk[: size - have]
        else:
            self._leftover = b""
        parts.append(chunk)
        have += len(chunk)
    return b"".join(parts)

HTTPStatusError(response)

Bases: Exception

A reply with an error status (4xx or 5xx). response is the reply; translate_errors() turns the error into the exception pathlib raises.

Source code in src/pathapi/http/_status.py

def __init__(self, response: HTTPResponse):
    super().__init__(response.status)
    self.response = response

response = response instance-attribute

HTTPTransport(*, ssl_context=None, verify=True, proxies=None)

Sends HTTP requests on the standard library: http.client over a socket, with TLS from ssl, and no third-party code.

send() makes exactly one request: it follows no redirect, adds no credentials and retries only as described below. A transport is a context manager, and close() closes every connection it holds.

https: servers are verified against ssl_context; without one, against ssl.create_default_context() (the system trust store). verify=False is the one way to turn verification off, and ssl_context wins over it.

proxies decides which proxy a request uses. None reads the environment (HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, NO_PROXY, through urllib.request.getproxies() and proxy_bypass()). Otherwise it is an object indexed with the request URL: it returns None (no proxy), one proxy, or an iterable of them to try in order when a connection cannot be made; KeyError means no proxy and the environment is not asked. A proxy is a URL or any object with a url attribute; a plain dict may use the keys requests uses ("http", "all", ...). Only http:// proxies are supported (CONNECT for https:), with Basic credentials from the proxy URL; any other scheme is OSError(EPROTONOSUPPORT). {} sends everything direct.

Each thread keeps one idle connection per server (at most 8), which a later request reuses unless the server has closed it or sent bytes on it since; the connections of a thread are closed when it ends. A request that fails before any response byte on a reused connection because the server had closed it is sent once more on a new connection, but only if it is repeatable (GET, HEAD, OPTIONS, PUT, DELETE, PROPFIND) and its body is None or in memory. Any other request reuses a connection only if it carried a response within the last second.

Source code in src/pathapi/http/_builtin.py

def __init__(
    self,
    *,
    ssl_context: "_ssl.SSLContext | None" = None,
    verify: bool = True,
    proxies: "_ty.Any" = None,
):
    self._ssl_context = ssl_context
    self._verify = bool(verify)
    if not self._verify and ssl_context is None:
        _log.warning(
            "HTTPTransport(verify=False): TLS certificates are not verified"
        )
    self._proxies = proxies
    self._pool = _pool.Pool()
    self._finalizer = _pool.finalize_with(self, self._pool)

close()

Closes every connection, idle or in use. Harmless when called again; a later request connects afresh.

Source code in src/pathapi/http/_builtin.py

def close(self) -> None:
    """Closes every connection, idle or in use. Harmless when called again;
    a later request connects afresh."""
    self._pool.close()

send(method, url, *, headers, body=None, timeout, stream=False)

One request, one response.

headers are sent as given; Accept-Encoding: gzip, deflate is added (and a coded body decoded) unless they name an Accept-Encoding. A body whose length is known without reading it goes under a Content-Length, any other chunked; a Content-Length in headers is used as it is. timeout is seconds for the connection and for each wait on the server: a number, a (connect, read) pair, or None to wait for ever. With stream=False the body has been read when this returns; with stream=True it is read from the response, which must be closed.

Raises ValueError before anything is sent for a header or URL with a control character, a URL with credentials or another scheme; TimeoutError, ConnectionError and ssl.SSLError for the network; OSError(EIO) for a reply that is not valid HTTP or ends early.

Source code in src/pathapi/http/_builtin.py

def send(
    self,
    method: str,
    url: str,
    *,
    headers: "_ty.Mapping[str, str]",
    body: "bytes | _ty.BinaryIO | _ty.Iterable[bytes] | None" = None,
    timeout: "float | tuple | None",
    stream: bool = False,
) -> HTTPResponse:
    """One request, one response.

    `headers` are sent as given; `Accept-Encoding: gzip, deflate` is added
    (and a coded body decoded) unless they name an `Accept-Encoding`.
    A `body` whose length is known without reading it goes under a
    `Content-Length`, any other chunked; a `Content-Length` in `headers`
    is used as it is. `timeout` is seconds for the connection and for each wait on the
    server: a number, a `(connect, read)` pair, or `None` to wait for ever.
    With `stream=False` the body has been read when this returns; with
    `stream=True` it is read from the response, which must be closed.

    Raises `ValueError` before anything is sent for a header or URL with
    a control character, a URL with credentials or another scheme;
    `TimeoutError`, `ConnectionError` and `ssl.SSLError` for the network;
    `OSError(EIO)` for a reply that is not valid HTTP or ends early."""
    method = check_method(method)
    target = check_url(url)
    pairs = check_headers(headers)
    connect_timeout, read_timeout = check_timeout(timeout)
    payload = _Payload(body, pairs)
    names = {name.lower() for name, _ in pairs}
    decode = wants_decoding(pairs)
    # As the server will see it: no fragment, "/" for an empty path.
    plain_url = f"{target.scheme}://{target.authority}{target.path}"
    route = proxies_for(self._proxies, plain_url, target.scheme, target.host)
    failure: "BaseException | None" = None
    for proxy in route:
        try:
            response = self._attempt(
                proxy,
                method,
                target,
                pairs,
                names,
                payload,
                connect_timeout,
                read_timeout,
                plain_url,
                decode,
            )
        except _CannotConnect as refused:
            failure = refused.error
            continue
        return response if stream else response._buffered()
    raise failure

Headers(pairs=())

Bases: Mapping

The headers of a response: a read-only mapping whose names compare without regard to case. A name sent more than once reads as its values joined with ", " (RFC 9110 5.3); get_all() returns them separately.

Source code in src/pathapi/http/_headers.py

def __init__(self, pairs: "_ty.Iterable[tuple[str, str]] | _abc.Mapping" = ()):
    if isinstance(pairs, _abc.Mapping):
        pairs = pairs.items()
    self._pairs = tuple((str(name), str(value)) for name, value in pairs)
    grouped: "dict[str, list[str]]" = {}
    self._spelling: "dict[str, str]" = {}
    for name, value in self._pairs:
        key = name.lower()
        grouped.setdefault(key, []).append(value)
        self._spelling.setdefault(key, name)
    self._joined = {key: ", ".join(values) for key, values in grouped.items()}

get_all(name)

Every value sent for name, in order.

Source code in src/pathapi/http/_headers.py

def get_all(self, name: str) -> "list[str]":
    """Every value sent for `name`, in order."""
    key = name.lower()
    return [value for field, value in self._pairs if field.lower() == key]

PreparedRequest

Bases: NamedTuple

A request that passed the checks of the transport contract.

headers hold text values, with the Content-Length of a body whose length is known (0 for a PATCH, POST or PUT without a body) and an Accept-Encoding of gzip, deflate unless the caller named one. body is None, bytes, an object that sends exactly the Content-Length bytes when it is read or iterated, or an iterable of bytes to send chunked. timeout is (connect, read). decode says whether the response decodes a gzip or deflate body: build it with HTTPResponse(..., decode=decode).

body instance-attribute

decode instance-attribute

headers instance-attribute

method instance-attribute

timeout instance-attribute

url instance-attribute

default_transport()

A new transport of the kind PATHAPI_HTTP_TRANSPORT names. A value that is empty, differently cased or unknown is ValueError naming the choices; one that another distribution provides and that is not installed is ImportError naming it.

Source code in src/pathapi/http/_transports.py

def default_transport():
    """A new transport of the kind `PATHAPI_HTTP_TRANSPORT` names. A value
    that is empty, differently cased or unknown is `ValueError` naming the
    choices; one that another distribution provides and that is not installed
    is `ImportError` naming it."""
    value = _os.environ.get(TRANSPORT_ENV)
    kind = _BUILTIN if value is None else value
    if kind == _BUILTIN:
        return HTTPTransport()
    try:
        transport_class = get_transport(kind)
    except ValueError:
        raise ValueError(
            f"{TRANSPORT_ENV}={kind!r} is not an HTTP transport "
            f"(expected one of {', '.join(map(repr, transport_names()))})"
        ) from None
    return transport_class()

drop_body(resp, limit=None)

Gives up the body of a reply nobody looks at: it is read, up to limit bytes (default MAX_REPLY_BYTES), so the connection can serve the next request, and resp is closed (which closes the connection when the body was longer). A failure while reading it is not the caller's concern: the status was already used.

Source code in src/pathapi/http/_status.py

def drop_body(resp: HTTPResponse, limit: "int | None" = None) -> None:
    """Gives up the body of a reply nobody looks at: it is read, up to `limit`
    bytes (default `MAX_REPLY_BYTES`), so the connection can serve the next
    request, and `resp` is closed (which closes the connection when the body
    was longer). A failure while reading it is not the caller's concern: the
    status was already used."""
    remaining = _settings.MAX_REPLY_BYTES if limit is None else limit
    try:
        while remaining > 0:
            chunk = resp.read(min(CHUNK, remaining))
            if not chunk:
                break
            remaining -= len(chunk)
    except (OSError, ValueError):
        pass
    finally:
        resp.close()

get_transport(name)

The class of the transport called name (case-sensitive), loading its entry point. builtin is always HTTPTransport: no entry point replaces it.

Raises ImportError naming what to install for a transport that another distribution provides and that is not installed, and ValueError naming the choices for any other name.

Source code in src/pathapi/http/_transports.py

def get_transport(name: str) -> type:
    """The class of the transport called `name` (case-sensitive), loading its
    entry point. `builtin` is always `HTTPTransport`: no entry point replaces
    it.

    Raises `ImportError` naming what to install for a transport that another
    distribution provides and that is not installed, and `ValueError` naming
    the choices for any other name."""
    if name == _BUILTIN:
        return HTTPTransport
    entry = _declared().get(name) if isinstance(name, str) else None
    if entry is not None:
        found = entry.load()
        if not callable(found):
            raise TypeError(f"the HTTP transport {name!r} is not a class")
        return found
    if isinstance(name, str) and name in _PROVIDED:
        distribution, package, install = _PROVIDED[name]
        raise ImportError(
            f"the {name!r} HTTP transport is provided by the {distribution} "
            f"distribution: {install}",
            name=package,
        )
    raise ValueError(
        f"{name!r} is not an HTTP transport "
        f"(expected one of {', '.join(map(repr, transport_names()))})"
    )

is_redirect(resp)

Source code in src/pathapi/http/_follow.py

def is_redirect(resp: HTTPResponse) -> bool:
    return resp.status in _REDIRECTS and "Location" in resp.headers

json_body(resp, what, limit=None)

The JSON document in the body of resp (at most limit bytes, default MAX_REPLY_BYTES, else OSError(EFBIG) naming what); ValueError when it is not one.

Source code in src/pathapi/http/_status.py

def json_body(resp: HTTPResponse, what: object, limit: "int | None" = None):
    """The JSON document in the body of `resp` (at most `limit` bytes, default
    `MAX_REPLY_BYTES`, else `OSError(EFBIG)` naming `what`); `ValueError` when
    it is not one."""
    return _json.loads(read_limited(resp, what, limit))

os_error(error_class, subject)

error_class built as pathlib builds it -- (errno, strerror, filename) -- for subject (a path or a URL, shown through str()), so e.errno and e.filename are set.

Source code in src/pathapi/http/_errors.py

def os_error(error_class, subject) -> OSError:
    """`error_class` built as pathlib builds it -- `(errno, strerror,
    filename)` -- for `subject` (a path or a URL, shown through `str()`), so
    `e.errno` and `e.filename` are set."""
    code = _ERRNOS.get(error_class, _errno.EIO)
    return error_class(code, _os.strerror(code), str(subject))

peer_text(text, limit=120)

text that came from the other end (a reason phrase, a Location, a header value) as an error message may show it: characters a terminal or a log would act on (controls, line and bidirectional separators) escaped, and cut to limit characters.

Source code in src/pathapi/http/_headers.py

def peer_text(text: object, limit: int = 120) -> str:
    """`text` that came from the other end (a reason phrase, a `Location`, a
    header value) as an error message may show it: characters a terminal or a
    log would act on (controls, line and bidirectional separators) escaped, and
    cut to `limit` characters."""
    text = str(text)
    cut = len(text) > limit
    shown = "".join(
        char if char.isprintable() else char.encode("unicode_escape").decode("ascii")
        for char in text[:limit]
    )
    return shown + "..." if cut else shown

prepare_request(method, url, *, headers, body=None, timeout)

The arguments of send() as the contract reads them.

method is upper-cased. Raises ValueError, before anything is sent, for a method or header name that is not an HTTP token, a header value or URL with a control character, a URL with credentials or another scheme, a Content-Length that is not a count of bytes or exceeds the body, and a timeout that is not a positive number, None or a pair of them; TypeError for a str body. None of the messages quotes the URL or a header value. The fragment is dropped from url.

Source code in src/pathapi/http/_transports.py

def prepare_request(
    method: str,
    url: str,
    *,
    headers: "_ty.Mapping[str, str] | None",
    body: "bytes | _ty.BinaryIO | _ty.Iterable[bytes] | None" = None,
    timeout: "float | tuple | None",
) -> PreparedRequest:
    """The arguments of `send()` as the contract reads them.

    `method` is upper-cased. Raises `ValueError`, before anything is sent, for
    a method or header name that is not an HTTP token, a header value or URL
    with a control character, a URL with credentials or another scheme, a
    `Content-Length` that is not a count of bytes or exceeds the body, and a
    `timeout` that is not a positive number, `None` or a pair of them;
    `TypeError` for a `str` body. None of the messages quotes the URL or a
    header value. The fragment is dropped from `url`."""
    method = _check_method(method)
    _check_url(url)
    _check_body(body)
    timeout = _check_timeout(timeout)
    pairs = _check_headers(headers)
    sent = dict(pairs)
    body, length = _prepare_body(body, pairs)
    if length is not None:
        sent["Content-Length"] = str(length)
    elif not _has_header(sent, "Content-Length"):
        if isinstance(body, (bytes, bytearray, memoryview)):
            sent["Content-Length"] = str(memoryview(body).nbytes)
        elif body is None and method in _EXPECTING_BODY:
            sent["Content-Length"] = "0"
    decode = _wants_decoding(sent.items())
    if not _has_header(sent, "Accept-Encoding"):
        sent["Accept-Encoding"] = "gzip, deflate"
    return PreparedRequest(method, url.split("#", 1)[0], sent, body, timeout, decode)

raise_for_status(resp)

Raises HTTPStatusError for a reply with status 400 or above.

Source code in src/pathapi/http/_status.py

def raise_for_status(resp: HTTPResponse) -> None:
    """Raises `HTTPStatusError` for a reply with status 400 or above."""
    if resp.status >= 400:
        raise HTTPStatusError(resp)

read_limited(resp, what, limit=None)

The decoded body of resp, which must be at most limit bytes (default MAX_REPLY_BYTES): OSError(EFBIG) naming what otherwise, with the connection closed.

Source code in src/pathapi/http/_status.py

def read_limited(resp: HTTPResponse, what: object, limit: "int | None" = None) -> bytes:
    """The decoded body of `resp`, which must be at most `limit` bytes (default
    `MAX_REPLY_BYTES`): `OSError(EFBIG)` naming `what` otherwise, with the
    connection closed."""
    limit = _settings.MAX_REPLY_BYTES if limit is None else limit
    resp._buffered(limit, too_large(what, limit))
    return resp.read()

server_size(text)

A byte count a server stated (Content-Length, getcontentlength) as an int, or None when it is not a plain non-negative decimal number.

Source code in src/pathapi/http/_errors.py

def server_size(text: "str | None") -> "int | None":
    """A byte count a server stated (`Content-Length`, `getcontentlength`) as
    an int, or `None` when it is not a plain non-negative decimal number."""
    text = (text or "").strip()
    if text.isascii() and text.isdigit() and len(text) <= 19:
        return int(text)
    return None

split_userinfo(url)

Split the userinfo out of an absolute URL: (url_without_userinfo, (user, password) | None), both parts percent-decoded to the octets they stand for. Credentials inside a request URL would end up in HTTPResponse.url, in error text and in redirect handling; they are sent in an Authorization header instead.

Source code in src/pathapi/http/_follow.py

def split_userinfo(url: str) -> "tuple[str, tuple[str|bytes, str|bytes] | None]":
    """Split the userinfo out of an absolute URL: `(url_without_userinfo,
    (user, password) | None)`, both parts percent-decoded to the octets they
    stand for. Credentials inside a request URL would end up in
    `HTTPResponse.url`, in error text and in redirect handling; they are sent
    in an `Authorization` header instead."""
    match = _RE_URL_SCHEME.match(url)
    if not match:
        return url, None
    start = match.end()
    end = len(url)
    for sep in "/?#":
        index = url.find(sep, start)
        if index != -1 and index < end:
            end = index
    userinfo, at, hostport = url[start:end].rpartition("@")
    if not at:
        return url, None
    user, _, password = userinfo.partition(":")
    auth = (credential(user), credential(password))
    return url[:start] + hostport + url[end:], (auth if any(auth) else None)

translate_errors(subject, *, conflict=FileExistsError)

Map a failed request to the exception pathlib raises for subject.

An HTTPStatusError becomes FileNotFoundError (404, 410), PermissionError (401, 403; 405 and 501 too, naming the status), conflict (409) or OSError(EIO) naming the status; a TimeoutError or ConnectionError from the transport is raised again with a message that holds subject and the exception's type name. conflict is what a 409 means: for a write (PUT, MKCOL, MOVE) RFC 4918 uses 409 for a missing parent collection, so writers pass FileNotFoundError. The cause is not chained: a traceback would otherwise print a URL, a header or a credential that the transport's own error carries.

Source code in src/pathapi/http/_errors.py

@_contextlib.contextmanager
def translate_errors(subject, *, conflict=FileExistsError):
    """Map a failed request to the exception `pathlib` raises for `subject`.

    An `HTTPStatusError` becomes `FileNotFoundError` (404, 410),
    `PermissionError` (401, 403; 405 and 501 too, naming the status),
    `conflict` (409) or `OSError(EIO)` naming the status; a `TimeoutError` or
    `ConnectionError` from the transport is raised again with a message that
    holds `subject` and the exception's type name. `conflict` is what a 409
    means: for a write (PUT, MKCOL, MOVE) RFC 4918 uses 409 for a missing
    parent collection, so writers pass `FileNotFoundError`. The cause is not
    chained: a traceback would otherwise print a URL, a header or a
    credential that the transport's own error carries."""
    try:
        yield
    except _HTTPStatusError as e:
        response = e.response
        status = response.status
        reason = peer_text(response.reason or "")
        if status in (404, 410):
            raise os_error(FileNotFoundError, subject) from None
        elif status in (401, 403):
            raise os_error(PermissionError, subject) from None
        elif status == 409:
            raise os_error(conflict, subject) from None
        elif status in (405, 501):
            raise PermissionError(
                _errno.EACCES,
                f"Method not allowed (HTTP {status})",
                str(subject),
            ) from None
        else:
            raise OSError(
                _errno.EIO,
                f"HTTP Error {status} {reason}".rstrip() + f" for {subject}",
            ) from None
    except TimeoutError as e:
        raise TimeoutError(f"Timeout for {subject} ({type(e).__name__})") from None
    except ConnectionError as e:
        raise ConnectionError(
            f"Connection error for {subject} ({type(e).__name__})"
        ) from None

transport_names()

The names of the HTTP transports, sorted: builtin, every entry of TRANSPORT_GROUP, and the transports that another distribution provides whether or not it is installed (requests). Nothing is imported.

Source code in src/pathapi/http/_transports.py

def transport_names() -> "tuple[str, ...]":
    """The names of the HTTP transports, sorted: `builtin`, every entry of
    `TRANSPORT_GROUP`, and the transports that another distribution provides
    whether or not it is installed (`requests`). Nothing is imported."""
    return tuple(sorted({_BUILTIN, *_declared(), *_PROVIDED}))