The transports

HttpxTransport implements the HTTP transport contract of pathapi.http: send() makes one request with an httpx.Client and returns an HTTPResponse, and close() closes a client the transport made. AsyncHttpxTransport implements the async contract: await send() makes one request with an httpx.AsyncClient on the running event loop and returns an AsyncHTTPResponse, and await aclose() closes a client the transport made. pathapi.http finds them through the httpx entries of its pathapi.http_transports and pathapi.http_async_transports entry-point groups, so PATHAPI_HTTP_TRANSPORT=httpx, PATHAPI_HTTP_ASYNC_TRANSPORT=httpx, pathapi.http.get_transport("httpx") and pathapi.http.get_async_transport("httpx") name them.

pathapi_httpx.HttpxTransport(client=None, **client_args)

Sends requests with an httpx.Client.

client is the client to use (the caller's: close() leaves it open), or None for a new one built from client_args (verify, cert, http2, proxy, mounts, limits, trust_env, auth, ...) that close() closes. client_args with a client is TypeError.

The transport sends one request and follows no redirect, whatever the client says. It sends exactly the headers it is given: the default headers and the cookies of a client are not added. The client's auth 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, ValueError or OSError(EIO); what httpx says is never passed on.

Source code in src/pathapi_httpx/_sync.py

def __init__(self, client: "_ty.Any" = None, **client_args):
    httpx = self._httpx = import_httpx()
    if client is not None:
        if client_args:
            raise TypeError(
                "client arguments apply to a client the transport makes"
            )
        if not callable(getattr(client, "send", None)):
            raise TypeError("a client is an object with a send() method")
    self._owns_client = client is None
    if client is None:
        client = owned_client(httpx.Client)(
            **{**client_args, "follow_redirects": False}
        )
        # Without a finalizer a client that is dropped leaves its sockets
        # to the garbage collector.
        self._finalizer = _weakref.finalize(self, client.close)
    self.client = client
    self._pin = Pin()

client = client instance-attribute

close()

Closes the client 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_httpx/_sync.py

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

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

One request, one response, as HTTPTransport.send().

Source code in src/pathapi_httpx/_sync.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()`."""
    prepared = prepare_request(
        method, url, headers=headers, body=body, timeout=timeout
    )
    request = build_request(self._httpx, prepared, prepared.body)
    auth = self._pin.auth(prepared, self._httpx.USE_CLIENT_DEFAULT)
    try:
        response = self.client.send(
            request, stream=True, follow_redirects=False, auth=auth
        )
    except Exception as error:
        raise failure(error) from None
    try:
        pairs = header_pairs(response)
    except BaseException:
        response.close()
        raise
    body = read_already(response)
    result = HTTPResponse(
        response.status_code,
        response.reason_phrase,
        pairs,
        requested_url(request),
        body if body is not None else _Body(self, response),
        decode=prepared.decode and body is None,
    )
    return result if stream else result.buffered()

pathapi_httpx.AsyncHttpxTransport(client=None, **client_args)

Sends requests with an httpx.AsyncClient, on the event loop that awaits them.

client is the client to use (the caller's: aclose() leaves it open), or None for a new one built from client_args (verify, cert, http2, proxy, mounts, limits, trust_env, auth, ...) that aclose() closes. client_args with a client is TypeError. Like any httpx.AsyncClient, it keeps connections of the loop it first sent on.

The transport sends one request and follows no redirect, whatever the client says. It sends exactly the headers it is given: the default headers and the cookies of a client are not added. The client's auth is used for the server of the first request only; a request to any other server carries just the Authorization it was given. A request body that is a file or an iterable is read on a worker thread.

Failures are TimeoutError, ConnectionError, ssl.SSLError, ValueError or OSError(EIO); what httpx says is never passed on.

Source code in src/pathapi_httpx/_async.py

def __init__(self, client: "_ty.Any" = None, **client_args):
    httpx = self._httpx = import_httpx()
    if client is not None:
        if client_args:
            raise TypeError(
                "client arguments apply to a client the transport makes"
            )
        if not _inspect.iscoroutinefunction(getattr(client, "send", None)):
            raise TypeError("a client is an object with an async send() method")
    self._owns_client = client is None
    if client is None:
        client = owned_client(httpx.AsyncClient)(
            **{**client_args, "follow_redirects": False}
        )
    self.client = client
    self._pin = Pin()

client = client instance-attribute

aclose() async

Closes the client if this transport made it. Harmless when called again.

Source code in src/pathapi_httpx/_async.py

async def aclose(self) -> None:
    """Closes the client if this transport made it. Harmless when called
    again."""
    if self._owns_client:
        await self.client.aclose()

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

One request, one response, as ThreadedTransport.send().

Source code in src/pathapi_httpx/_async.py

async 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,
) -> AsyncHTTPResponse:
    """One request, one response, as `ThreadedTransport.send()`."""
    prepared = prepare_request(
        method, url, headers=headers, body=body, timeout=timeout
    )
    content = prepared.body
    if content is not None and not isinstance(content, (bytes, bytearray)):
        content = _pieces(content)
    request = build_request(self._httpx, prepared, content)
    auth = self._pin.auth(prepared, self._httpx.USE_CLIENT_DEFAULT)
    try:
        response = await self.client.send(
            request, stream=True, follow_redirects=False, auth=auth
        )
    except Exception as error:
        raise failure(error) from None
    try:
        pairs = header_pairs(response)
    except BaseException:
        await response.aclose()
        raise
    body = read_already(response)
    result = AsyncHTTPResponse(
        response.status_code,
        response.reason_phrase,
        pairs,
        requested_url(request),
        body if body is not None else _Body(self, response),
        decode=prepared.decode and body is None,
    )
    return result if stream else await result.buffered()