Authentication, transports and lifecycle

Authentication belongs to its implementation

HttpClient(auth=...) accepts the exported structural HttpAuth interface. The authenticator owns credentials, signing rules, expiry, refresh coordination, token storage and its handshake step budget. HTTP sends messages, supplies response evidence and releases exchange resources. No PostgreSQL, TOTP or browser dependencies are imported by this integration.

A basic fixed-token adapter can be an ordinary Python object:

from collections.abc import AsyncGenerator
from pydantic import SecretStr
from common_http import HttpClient, HttpResponseHead, Request


class BearerAuth:
    def __init__(self, token: str):
        self._token = SecretStr(token)

    async def flow(self, request: Request) -> AsyncGenerator[Request, HttpResponseHead]:
        headers = request.headers.replace(
            "Authorization", [f"Bearer {self._token.get_secret_value()}"]
        )
        yield request.model_copy(update={"headers": headers})

    def requires_body(self, response: HttpResponseHead) -> bool:
        return False


http = HttpClient.create(base_url="https://provider.example/", auth=BearerAuth("example-token"))

This example intentionally has no refresh algorithm. The shared common-auth implementation can implement the same interface without a different HTTP client.

Multi-step flows and signing

The flow yields a Request and receives its response through asend. Ending the flow accepts that response. Yielding another request authorizes another exchange. There is no hardcoded single-401 retry or implicit refresh in HTTP.

requires_body(head) selects responses that must be buffered before the flow continues. Those arrive as HttpResponse with retained bytes, or empty-response semantics when no bytes exist. Buffering obeys max_response_bytes, including when the caller uses streaming. A fresh flow is created per attempt and followed original-origin redirect; concurrent flows need independent handshake state.

Before authentication, transport.prepare freezes the canonical target, headers and encoded body bytes, including multipart boundaries. Auth yields final binary bodies and canonical URLs. This lets body signatures cover bytes actually sent. Authentication follow-ups stay on the original origin. The adapter owns its step limit and shared refresh synchronization; buffered operations also have their configured per-attempt timeout. Refresh/token calls should use a separate unauthenticated client to avoid recursion and pool contention.

Per-exchange admission

async def admit(request: Request) -> None:
    await limiter.acquire(account="provider-account", cost=1)


http = HttpClient.create(base_url="https://provider.example/", admit=admit)

limiter is supplied by the application or future shared rate package. The hook runs before every physical exchange, including retries, redirects and auth follow-ups. A hook may wait or reject by raising. It does not imply a built-in distributed backend. Separate token-refresh clients need their own admission.

Cookie persistence defaults on. Domain, path, secure and expiry rules control attachment; an explicit Cookie field takes precedence. Set persist_cookies=False to disable the client’s jar. The jar is in memory and cleared when the client closes.

Redirects default off. With follow_redirects=True, max_redirects bounds the chain. POST 301/302 and non-HEAD 303 become GET without a body. 307/308 preserve method/body. Cross-origin redirects forward only Accept, Accept-Language and User-Agent from carried headers, rebuild destination cookies, and disable the original authenticator. Automatic HTTPS downgrade and cross-origin body replay are rejected. Use an explicit new request when a provider requires such a step and the consumer has decided which credentials/body may be sent.

Stream response bytes

import asyncio
from common_http import HttpClient

async with HttpClient.create(base_url="https://provider.example/") as http:
    request = http.build_request("/download")
    async with asyncio.timeout(120):
        async with http.stream(request) as (head, chunks):
            if not head.is_success:
                raise RuntimeError(f"Download failed: HTTP {head.status}")
            async for chunk in chunks:
                await sink.write(chunk)

The sink is application-owned. Chunks are HTTP content-decoded bytes. Exiting the context early or cancellation closes the response. Streaming does not create a buffered HttpResult, run automatic retries or apply runner batch/pacing settings. Phase timeouts and per-exchange admission still apply. The client does not enforce the buffered response cap on an ordinary stream; the sink controls total size.

Owned and borrowed resources

The default HTTPX transport is owned by HttpClient and closed with it. Injected transports and authenticators are borrowed. Each client/transport belongs to one asyncio loop and closed instances cannot reopen. Close rejects active logical operations, including retry waits and batches; finish or cancel and await those operations before closing.

import httpx
from common_http import HttpClient, HttpxTransport

async with httpx.AsyncClient() as engine:
    transport = HttpxTransport(engine=engine)
    try:
        async with HttpClient(transport=transport) as http:
            result = await http.get("https://provider.example/health")
    finally:
        await transport.aclose()

Configure pool/TLS/proxy/protocol settings on a borrowed transport/engine itself. Explicit creation overrides are rejected when borrowing. HTTPX engine default headers, query, cookies and auth are not merged into prepared messages. A borrowed engine may still store received cookies in its own jar. Custom native transports must avoid hidden retries if the caller expects one execution owner.

BaseTransport defines lifecycle and one managed exchange. Implement its native open/close/exchange/capability seams for another engine; override prepare when it serializes structured bodies. It never invokes execution runners or batching. See the transport reference.