The transport

RequestsTransport implements the HTTP transport contract of pathapi.http: send() makes one request with a requests.Session and returns an HTTPResponse, and close() closes a session the transport made. pathapi.http finds it through the requests entry of its pathapi.http_transports entry-point group, so PATHAPI_HTTP_TRANSPORT=requests and pathapi.http.get_transport("requests") name it.

pathapi_requests.RequestsTransport(session=None, *, proxies=None, **requests_args)

Sends requests with a requests.Session.

session is the session to use (the caller's: close() leaves it open), or None for a new one that close() closes. proxies is a provider indexed with the request URL, as the built-in transport takes one: the answer is a proxy (a URL, or an object with a url), an iterable of them of which only the first is used, or None for no proxy; KeyError means no opinion, and then requests decides from the session and the environment, as it does when proxies is None. A plain dict may use the keys requests does. requests_args (verify, cert, auth, ...) are passed to every session.request().

The transport sends one request and follows no redirect. What the session holds for authentication (auth, an Authorization among its headers, ~/.netrc) is used for the server of the first request only; a request to any other server carries just the Authorization it was given.

Failures are TimeoutError, ConnectionError, ssl.SSLError, or OSError(EIO); the text of a requests exception, which can quote a proxy URL with credentials, is never passed on.

Source code in src/pathapi_requests/_transport.py

def __init__(
    self, session: "_ty.Any" = None, *, proxies: "_ty.Any" = None, **requests_args
):
    self._requests = _import_requests()
    if session is not None and not callable(getattr(session, "request", None)):
        raise TypeError("a session is an object with a request() method")
    self._owns_session = session is None
    self.session = session if session is not None else self._requests.Session()
    if self._owns_session:
        # Without a finalizer an old `urllib3` leaves the sockets of a
        # dropped session to the garbage collector.
        self._finalizer = _weakref.finalize(self, self.session.close)
    self._provider = proxies
    self._args = requests_args
    self._pinned: "tuple | None" = None
    self._lock = _threading.Lock()

session = session if session is not None else self._requests.Session() instance-attribute

close()

Closes the session if this transport made it, and does so when the transport is collected without having been closed. Harmless when called again.

Source code in src/pathapi_requests/_transport.py

def close(self) -> None:
    """Closes the session if this transport made it, and does so when the
    transport is collected without having been closed. Harmless when
    called again."""
    if self._owns_session:
        self._finalizer.detach()
        self.session.close()

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

One request, one response, as HTTPTransport.send(); requests' own trust store and authentication apply.

Source code in src/pathapi_requests/_transport.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, as `HTTPTransport.send()`; `requests`'
    own trust store and authentication apply."""
    request = prepare_request(
        method, url, headers=headers, body=body, timeout=timeout
    )
    sent = request.headers
    args = {k: v for k, v in self._args.items() if k not in _OWN_KEYS}
    ambient = self._args.get("headers")
    with self._lock:
        if self._pinned is None:
            self._pinned = _origin(request.url)
        home = self._pinned == _origin(request.url)
    if ambient:
        sent = _merge_headers(
            ambient if home else _drop_header(ambient, "Authorization"), sent
        )
    explicit = next(
        (v for k, v in sent.items() if k.lower() == "authorization"), None
    )
    if not home:
        args["auth"] = _OnlyThis(explicit)
    elif (
        explicit is not None
        and args.get("auth") is None
        and not getattr(self.session, "auth", None)
    ):
        args["auth"] = _OnlyThis(explicit)
    proxies = self._proxies_for(request.url)
    if proxies is not None:
        args["proxies"] = proxies
    try:
        response = self.session.request(
            method=request.method,
            url=request.url,
            headers=sent,
            data=request.body,
            timeout=request.timeout,
            stream=True,
            allow_redirects=False,
            **args,
        )
    except self._requests.exceptions.RequestException as error:
        raise self._failure(error) from None
    except ValueError as error:
        # The URL and the headers were checked before this call, so this
        # is `requests` reading what the server said: a `Location` it
        # cannot parse, when it looks ahead to the redirect.
        raise OSError(
            _errno.EIO, f"the reply could not be read ({type(error).__name__})"
        ) from None
    status = response.status_code
    if status < 200:
        # An interim reply taken as the final one (urllib3 skips only
        # `100`); the real reply is still to come on that connection.
        response.close()
        raise OSError(
            _errno.EIO,
            f"the server's reply was an interim one (HTTP {status}), "
            "not a final one",
        )
    length = response.headers.get("Content-Length", "")
    if request.method == "HEAD" or status in (204, 304):
        expected = 0
    elif "chunked" in response.headers.get("Transfer-Encoding", "").lower():
        expected = None
    else:
        expected = int(length) if length.isascii() and length.isdigit() else None
    result = HTTPResponse(
        response.status_code,
        response.reason,
        _header_pairs(response),
        getattr(response, "url", None) or request.url,
        _Body(self, response, expected),
        decode=request.decode,
    )
    return result if stream else result.buffered()