API Reference

Protocols

`HTTPClientProtocol`

Protocol for HTTP client implementations.

start

async def start() -> None

source

Start the HTTP client and its underlying connection pool.

stop

async def stop() -> None

source

Stop the HTTP client and release all connection resources.

request

async def request(
    method: str,
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform an arbitrary HTTP request.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (GET, POST, PUT, …).
`url`	str	Request URL. **kwargs: Additional options (headers, data, json, params, …).

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

get

async def get(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform GET request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

post

async def post(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform POST request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options (e.g. ``json=,` `data=``).

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

put

async def put(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform PUT request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

delete

async def delete(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform DELETE request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

patch

async def patch(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform PATCH request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

head

async def head(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform HEAD request.

Parameters

Parameter	Type	Description
`url`	str	Request URL. **kwargs: Additional options.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

stream_response

def stream_response(
    method: str,
    url: str,
    **kwargs: Any
) -> AbstractAsyncContextManager[Any]

source

Open a raw byte stream with response status and headers.

The returned async context manager owns the connection. Its value exposes status, headers and aiter_bytes(). This contract lets gateways classify an upstream response before forwarding a long-lived stream without buffering it.

`InterceptorChainProtocol`

source

Protocol for interceptor chain management.

Manages a collection of interceptors and orchestrates their execution in sequence. Each interceptor in the chain processes the request/response before passing control to the next interceptor.

Example

class InterceptorChain:
    def __init__(self, interceptors: list[InterceptorProtocol]):
        self._interceptors = interceptors

    async def execute_request(self, context: Any) -> Any:
        for interceptor in self._interceptors:
            context = await interceptor.intercept_request(context)
        return context

    async def execute_response(self, response: Any) -> Any:
        for interceptor in reversed(self._interceptors):
            response = await interceptor.intercept_response(response)
        return response

add_interceptor

def add_interceptor(interceptor: InterceptorProtocol) -> None

source

Add an interceptor to the chain.

Parameters

Parameter	Type	Description
`interceptor`	InterceptorProtocol	The interceptor to add.

remove_interceptor

def remove_interceptor(interceptor: InterceptorProtocol) -> None

source

Remove an interceptor from the chain.

Parameters

Parameter	Type	Description
`interceptor`	InterceptorProtocol	The interceptor to remove.

process_request

async def process_request(context: Any) -> Any

source

Process request through the interceptor chain.

Parameters

Parameter	Type	Description
`context`	Any	Request context to process.

Returns

Type	Description
Any	Modified request context after all interceptors.

process_response

async def process_response(response: Any) -> Any

source

Process response through the interceptor chain.

Parameters

Parameter	Type	Description
`response`	Any	Response to process.

Returns

Type	Description
Any	Modified response after all interceptors.

`InterceptorProtocol`

source

Protocol for request/response interceptors.

Interceptors are applied to every request/response cycle in HTTPClient. Implementations receive typed RequestContext objects and may modify them before the request is dispatched, or inspect / annotate the raw response after it arrives.

Example

class LoggingInterceptor: async def intercept_request(self, context: Any) -> Any: logger.info(“outbound_request”, method=context.method, url=context.url) return context

async def intercept_response(self, response: Any) -> Any:
    logger.info("inbound_response", status=response.status)
    return response

intercept_request

async def intercept_request(context: Any) -> Any

source

Called before a request is dispatched.

Parameters

Parameter	Type	Description
`context`	Any	RequestContext for the outbound request. Implementations may mutate and return it.

Returns

Type	Description
Any	The (possibly modified) request context.

intercept_response

async def intercept_response(response: Any) -> Any

source

Called after a response is received from the server.

Parameters

Parameter	Type	Description
`response`	Any	Raw response object from the underlying HTTP library. Implementations may annotate or replace it.

Returns

Type	Description
Any	The (possibly modified) response.

Classes

`BaseURLHTTPClient`

source

Async HTTP client with base-URL resolution and context-manager lifecycle.

Wraps HTTPClient and adds:

base_url prepended to every relative request path.
Default headers merged into every request.
Direct HttpResponse returns from verb methods (raises on infrastructure failures rather than returning Result).
stream async context manager for line-oriented streaming responses (SSE, NDJSON).
Lazy initialisation — no explicit start() call required; the underlying connection pool is created on first request and torn down by close.

Parameters

Parameter	Type	Description
`base_url`		Base URL prepended to every relative path.
`headers`		Default request headers merged with per-request headers.
`timeout`		Request timeout in seconds (default 300).
`name`		Logical name used in log messages.
`retry`		Optional retry policy injected into the underlying HTTPClient.
`circuit_breaker`		Optional circuit breaker injected into the underlying HTTPClient.
`enforce_url_safety`		Keep the SSRF gate on (default). Set ``False`` only when every target is a fixed, operator-configured base URL and never a caller-supplied URL, for example a local sandbox.

Example

async with BaseURLHTTPClient(
    base_url="https://api.example.com/v1",
    headers={"Authorization": "Bearer token"},
    timeout=60.0,
) as client:
    resp = await client.post("/completions", json=payload)
    resp.raise_for_status()

async with BaseURLHTTPClient(
    base_url="https://api.example.com/v1",
    headers={"Authorization": "Bearer token"},
    timeout=60.0,
) as client:
    resp = await client.post("/completions", json=payload)
    resp.raise_for_status()

__init__

def __init__(
    base_url: str = '',
    headers: dict[str, str] | None = None,
    timeout: float = 300.0,
    name: str = 'base-url-http-client',
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None,
    enforce_url_safety: bool = True
) -> None

source

async def close() -> None

source

Stop the underlying HTTP client and release all connections.

stop

async def stop() -> None

source

Alias for close.

request

async def request(
    method: str,
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Make an HTTP request, returning HttpResponse directly.

Infrastructure failures are raised as exceptions (not wrapped in Result). HTTP 4xx/5xx responses are returned as-is; call raise_for_status to convert them to exceptions when desired.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (GET, POST, …).
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp (``json=,` `data=,` `headers=``, …).

Returns

Type	Description
HttpResponse	Framework HttpResponse.

get

async def get(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform a GET request.

Parameters

Parameter	Type	Description
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp.

Returns

Type	Description
HttpResponse	Framework HttpResponse.

post

async def post(
    url: str,
    data: Any = None,
    **kwargs: Any
) -> HttpResponse

source

Perform a POST request.

Parameters

Parameter	Type	Description
`url`	str	Relative path or absolute URL.
`data`	Any	Optional body. ``dict` `is sent as JSON; other types as raw` `data. Use` `json=`` kwarg for explicit JSON. **kwargs: Forwarded to aiohttp.

Returns

Type	Description
HttpResponse	Framework HttpResponse.

put

async def put(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform a PUT request.

Parameters

Parameter	Type	Description
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp.

Returns

Type	Description
HttpResponse	Framework HttpResponse.

delete

async def delete(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform a DELETE request.

Parameters

Parameter	Type	Description
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp.

Returns

Type	Description
HttpResponse	Framework HttpResponse.

patch

async def patch(
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Perform a PATCH request.

Parameters

Parameter	Type	Description
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp.

Returns

Type	Description
HttpResponse	Framework HttpResponse.

stream_response

async def stream_response(
    method: str,
    url: str,
    **kwargs: Any
) -> AsyncIterator[StreamResponse]

source

Stream a relative URL with status, headers and raw body chunks.

This is the base-URL counterpart to quadkit.http.client.HTTPClient.stream_response. It keeps the response metadata visible to protocol gateways while preserving the client’s default URL and header policy.

stream

async def stream(
    method: str,
    url: str,
    **kwargs: Any
) -> AsyncIterator[StreamContext]

source

Async context manager for line-oriented streaming responses.

Opens a persistent HTTP connection and wraps the response in a StreamContext that exposes aiter_lines for consuming SSE / NDJSON streams — the dominant format used by LLM provider streaming APIs.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (typically ``"POST"`` for LLM completions).
`url`	str	Relative path or absolute URL. **kwargs: Forwarded to aiohttp (``json=,` `data=,` `headers=``, …).

Yields

Type	Description
AsyncIterator[StreamContext]	StreamContext with status, raise_for_status, and aiter_lines.

Example

async with client.stream("POST", "/chat/completions", json=payload) as resp:
    resp.raise_for_status()
    async for line in resp.aiter_lines():
        if line.startswith("data: "):
            ...

async with client.stream("POST", "/chat/completions", json=payload) as resp:
    resp.raise_for_status()
    async for line in resp.aiter_lines():
        if line.startswith("data: "):
            ...

`ConnectionPool`

source

Async HTTP connection pool backed by ``aiohttp``.

Parameters

Parameter	Type	Description
`max_connections`		Maximum total concurrent connections (default: 10).
`max_keepalive_connections`		Keep-alive connections per host (default: 5).
`timeout`		Request timeout in seconds (default: 30.0).
`ttl_dns_cache`		DNS cache TTL in seconds (default: 300).
`force_close`		Force-close connections after each use (default: False).
`verify_ssl`		Verify SSL certificates (default: True). Set to ``False`` only in trusted internal networks — never in production for external endpoints.

Example

pool = ConnectionPool(max_connections=100, timeout=60.0)
await pool.start()
response = await pool._session.get("https://api.example.com")
await pool.stop()

pool = ConnectionPool(max_connections=100, timeout=60.0)
await pool.start()
response = await pool._session.get("https://api.example.com")
await pool.stop()

__init__

def __init__(
    max_connections: int = 10,
    max_keepalive_connections: int = 5,
    max_connections_per_host: int = 10,
    timeout: float = 30.0,
    ttl_dns_cache: int = 300,
    force_close: bool = False,
    verify_ssl: bool = True,
    proxy: str | None = None,
    trust_env: bool = True,
    cookie_jar: bool = True
) -> None

source

start

async def start() -> None

source

Initialise the connection pool and create the aiohttp session.

stop

async def stop() -> None

source

Close the connection pool and release all resources.

is_started

def is_started() -> bool

source

Return True if the pool session is active.

pin_dns

def pin_dns(
    host: str,
    addresses: Sequence[ipaddress.IPv4Address | ipaddress.IPv6Address]
) -> None

source

Pin the safety-validated addresses for host.

The connector resolves host exclusively to the pinned set until it is pinned again, so the address actually connected to is one of the addresses the SSRF gate approved (DNS-rebinding mitigation).

Safe to call before start — the pin is applied to the resolver when the session is created.

session

property session() -> HTTPSessionProtocol | None

source

The underlying aiohttp session (None before start).

warm_up

async def warm_up(urls: list[str]) -> None

source

Pre-establish connections to known endpoints.

Fires a lightweight HEAD (falling back to GET) request to each URL so that the TCP and TLS handshake has already been completed before the first real request arrives. Errors are swallowed silently — this is a best-effort optimisation, not a connectivity check.

Parameters

Parameter	Type	Description
`urls`	list[str]	Base URLs to pre-connect to (e.g. ``["https://api.example.com"]``).

Example

await pool.start()
await pool.warm_up(["https://api.example.com", "https://cdn.example.com"])

await pool.start()
await pool.warm_up(["https://api.example.com", "https://cdn.example.com"])

`ConnectionPoolConfig`

source

Configuration for the HTTP connection pool.

Attributes: max_connections: Maximum total concurrent connections (default: 10). max_keepalive_connections: Keep-alive connections per host (default: 5). max_connections_per_host: Maximum connections per individual host (default: 10). timeout: Request timeout in seconds (default: 30.0). ttl_dns_cache: DNS cache TTL in seconds (default: 300). force_close: Force-close connections after use (default: False).

`HTTPClient`

source

Async HTTP client with connection pooling and optional resilience.

All dependencies are provided at construction time (constructor injection). When used through HTTPProvider, retry_policy and circuit_breaker are resolved from the container and injected here. For standalone usage, pass them directly or accept the defaults.

Parameters

Parameter	Type	Description
`config`		Pool and client configuration. Defaults to HTTPClientConfig with framework defaults.
`pool`		Optional pre-configured ConnectionPool. Created from config when not provided.
`retry_policy`		Retry policy for transient failures. Defaults to a RetryPolicy with framework defaults.
`circuit_breaker`		Optional circuit breaker; disabled when ``None``.
`interceptors`		Zero or more interceptors applied to every request.
`resilience`		Optional ResiliencePipelineProtocol that combines retry, circuit-breaker, bulkhead and timeout into a single composable pipeline. When provided it takes precedence over the individual ``retry_policy` `and` `circuit_breaker`` parameters.
`metrics`		Optional MetricsRecorderProtocol for emitting ``http.request.duration` `(histogram) and` `http.request.status`` (counter) per-request metrics.

Example

config = HTTPClientConfig()
async with HTTPClient.session_context(config) as client:
response = await client.get("https://api.example.com/data")

__init__

def __init__(
    config: HTTPClientConfig | None = None,
    pool: ConnectionPool | None = None,
    retry_policy: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None,
    interceptors: Iterable[InterceptorProtocol] = (),
    resilience: ResiliencePipelineProtocol | None = None,
    metrics: MetricsRecorderProtocol | None = None
) -> None

source

config

property config() -> HTTPClientConfig

source

Return the client configuration.

pool

property pool() -> ConnectionPool

source

Return the underlying connection pool.

pool

def pool(value: ConnectionPool) -> None

source

Replace the connection pool (mainly for testing).

start

async def start() -> None

source

Start the HTTP client and its connection pool.

stop

async def stop() -> None

source

Stop the HTTP client and close the connection pool.

request

async def request(
    method: str,
    url: str,
    **kwargs: Any
) -> HttpResponse

source

Make an HTTP request with retry and optional circuit-breaker protection.

Note Infrastructure-level method — this method raises exceptions for all failures (circuit open, retries exhausted, connection errors). Use the high-level verb methods (get, post, etc.) in application/domain code — they return Result and avoid exception-based control flow for expected failures.

Summary of the two-layer strategy:

* ``request()`` — infrastructure layer, raises exceptions.
* ``get()``, ``post()``, etc. — domain layer, return ``Result``.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (``GET,` `POST``, etc.)
`url`	str	Target URL. **kwargs: Additional keyword arguments forwarded to ``aiohttp``.

Returns

Type	Description
HttpResponse	Framework-owned HttpResponse.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.
HTTPConnectionError	When the connection cannot be established.
HTTPTimeoutError	When the request times out.
HTTPUnsafeURLError	When the target is unsafe or a proxy would bypass URL-safety enforcement.

`HTTPClientConfig`

source

Top-level configuration for HTTPClient.

Attributes: pool: Connection pool settings. proxy: Optional HTTP/HTTPS proxy URL (e.g. "http://proxy.example.com:8080"). Applied to requests only when URL-safety enforcement is disabled; proxies resolve destination hosts remotely and bypass the DNS pin. trust_env: Whether to read proxy settings from environment variables (HTTP_PROXY, HTTPS_PROXY, NO_PROXY). Defaults to True when URL safety is disabled. Enabling URL safety suppresses env proxies because their remote DNS resolution bypasses DNS pins. cookie_jar: Whether to enable an in-memory cookie jar that persists cookies across requests within a single HTTPClient session. Defaults to True. enforce_url_safety: Reject per-request URLs that could reach private/reserved hosts. Defaults to True (safe by default); set False only for clients whose targets are fixed, trusted, operator-set URLs.

`HTTPModule`

source

Async outbound HTTP client backed by aiohttp with connection pooling.

Call configure to configure the HTTP client with custom settings and optional resilience integration (retry, circuit breaker).

Usage (defaults)

@module(imports=[HTTPModule.configure()])
class AppModule(Module):
    pass

@module(imports=[HTTPModule.configure()])
class AppModule(Module):
    pass

Usage (configured)

from quadkit.http.config import HTTPClientConfig

@module(
    imports=[HTTPModule.configure(HTTPClientConfig(timeout=30))]
)
class AppModule(Module):
    pass

from quadkit.http.config import HTTPClientConfig

@module(
    imports=[HTTPModule.configure(HTTPClientConfig(timeout=30))]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: Any | None = None
) -> DynamicModule

source

Create an HTTPModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	Any \| None	HTTPClientConfig or ``None`` for framework defaults.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return a no-op HTTPModule for unit testing.

Registers an HTTP client with default configuration. No real outbound connections are made unless explicitly called.

Returns

Type	Description
DynamicModule	A DynamicModule with default HTTP client configuration.

`HTTPProvider`

source

Framework provider for the HTTP client.

Registers HTTPClient (and its HTTPClientProtocol binding) as a container-scoped singleton. Resilience dependencies are resolved from the container during boot and injected into the client — no service locator is used.

Parameters

Parameter	Type	Description
`config`		Optional HTTP client configuration. Framework defaults apply when not provided.

__init__

def __init__(config: HTTPClientConfig | dict[str, Any] | None = None) -> None

source

from_config

def from_config(
    cls,
    config: HTTPClientConfig,
    **context: Any
) -> Self

source

Create provider from config object.

async def register(container: ContainerRegistrarProtocol) -> None

source

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The DI registrar.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Construct and start the HTTP client with injected resilience.

Resolves RetryPolicyProtocol and optionally CircuitBreakerRegistryProtocol from the container. Falls back to framework defaults when the container has no binding for them.

Parameters

Parameter	Type	Description
`container`	ContainerResolverProtocol	The DI resolver.

shutdown

async def shutdown() -> None

source

Stop and dispose the HTTP client.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return the health of the HTTP provider.

Parameters

Parameter	Type	Description
`timeout`	float	Unused; retained for interface compatibility.

Returns

Type	Description
HealthCheckResult	HealthCheckResult reflecting whether the client is active.

`HTTPRequestSentHook`

source

Payload fired when an outbound HTTP request is dispatched.

Attributes: method: HTTP verb in upper-case (e.g. "GET"). url: Full URL of the outbound request.

`HTTPResponseReceivedHook`

source

Payload fired when an outbound HTTP response is received.

Attributes: method: HTTP verb of the originating request. url: URL that was requested. status_code: HTTP status code of the received response.

`RequestCompletedEvent`

source

HTTP request completed successfully.

Consumed by: request metrics, audit logging, performance analysis.

`RequestContext`

source

Context information for an outbound HTTP request.

Used by interceptors to inspect and modify the request before it is sent.

Attributes: method: HTTP method (upper-cased). url: Target URL. headers: Request headers as a plain dict. service_name: Logical service name if known (service-mesh use cases). attempt: Current retry attempt number (0-indexed). start_time: Timestamp when the request was initiated.

`RequestRetryExhaustedEvent`

source

HTTP request exhausted all retry attempts.

Consumed by: retry management, circuit breaker logic, error reporting.

`RequestTimeoutEvent`

source

HTTP request timed out before completion.

Consumed by: timeout tracking, performance monitoring, alerting.

`ResponseContext`

source

Context information for a received HTTP response.

Used by interceptors to inspect response metadata.

Attributes: status: HTTP status code. headers: Response headers. content_length: Optional response body length in bytes. duration: Round-trip duration in seconds. success: True when status < 400. error: Human-readable error message when the request failed.

`StreamContext`

source

Wraps an aiohttp response for line-oriented streaming.

Exposes status, raise_for_status(), and aiter_lines() without leaking the underlying aiohttp type to callers.

Parameters

Parameter	Type	Description
`resp`		The raw aiohttp ClientResponse, owned by the enclosing context manager for the duration of the stream.

__init__

def __init__(resp: aiohttp.ClientResponse) -> None

source

status

property status() -> int

source

HTTP status code of the response.

raise_for_status

def raise_for_status() -> None

source

Raise HttpStatusError for 4xx/5xx.

Raises

Exception	Description
HttpStatusError	When the status code is 400 or higher.

aiter_lines

async def aiter_lines() -> AsyncIterator[str]

source

Yield decoded text lines from the streaming response body.

Yields

Type	Description
AsyncIterator[str]	Each line decoded from UTF-8 with trailing CR/LF stripped.

`StreamResponse`

source

Metadata and body iterator for an open HTTP response stream.

HTTPClient.stream historically exposed only the body iterator. Relay gateways also need the upstream status and headers before forwarding the first byte, especially to classify a 401/429/5xx without buffering a successful SSE response. The context manager that created this object owns the underlying connection; consume aiter_bytes while it is open.

aiter_bytes

async def aiter_bytes() -> AsyncIterator[bytes]

source

Yield raw response bytes as they arrive.

`WebSocketClient`

source

Async context-managed outbound WebSocket client.

__init__

def __init__(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    connect_timeout: float = 30.0,
    heartbeat: float | None = 30.0
) -> None

source

closed

property closed() -> bool

source

Whether the upstream connection has closed.

send_json

async def send_json(payload: Any) -> None

source

Send one JSON event.

send_text

async def send_text(payload: str) -> None

source

Send one text frame.

send_bytes

async def send_bytes(payload: bytes) -> None

source

Send one binary frame.

receive

async def receive() -> WebSocketMessage

source

Receive and normalize one upstream frame.

async def close() -> None

source

Close the WebSocket and its HTTP session.

`WebSocketMessage`

source

A normalized message received from an outbound WebSocket.

Functions

`build_url`

build_url

def build_url(
    base: str,
    path: str = '',
    params: dict[str, Any] | None = None
) -> str

source

Build a complete URL from base, path, and query parameters.

Parameters

Parameter	Type	Description
`base`	str	Base URL (e.g. ``"https://api.example.com"``).
`path`	str	URL path to append (e.g. ``"/users/123"``).
`params`	dict[str, Any] \| None	Optional query parameters; ``None`` values are omitted.

Returns

Type	Description
str	Complete URL with path and encoded query string.

Example

build_url("https://api.example.com", "/users", {"page": 1})

'https://api.example.com/users?page=1'

`extract_json_type`

extract_json_type

def extract_json_type(content_type: str) -> str | None

source

Extract the JSON MIME type from a ``Content-Type`` header value.

Parameters

Parameter	Type	Description
`content_type`	str	Raw ``Content-Type`` header value.

Returns

Type	Description
str \| None	The base MIME type if it contains ``"json", otherwise` `None``.

Example

extract_json_type("application/json; charset=utf-8")

'application/json'
>>> extract_json_type("text/html")
None

`format_timeout`

format_timeout

def format_timeout(timeout: float | None) -> str

source

Format a timeout value for human-readable display.

Parameters

Parameter	Type	Description
`timeout`	float \| None	Timeout in seconds, or ``None`` for no timeout.

Returns

Type	Description
str	Formatted string such as ``"30.0s"` `or` `"no timeout"``.

`merge_headers`

merge_headers

def merge_headers(
    *header_dicts: Mapping[str, Any],
    normalize: bool = True
) -> dict[str, str]

source

Merge multiple header mappings; later entries win on duplicate keys.

Parameters

Parameter	Type	Description
`normalize`	bool	When ``True``, lowercase keys and strip values (default).

Returns

Type	Description
dict[str, str]	Merged ``dict[str, str]``.

`parse_headers`

parse_headers

def parse_headers(headers: Mapping[str, Any]) -> dict[str, str]

source

Normalise HTTP headers: lowercase keys, strip whitespace from values.

Parameters

Parameter	Type	Description
`headers`	Mapping[str, Any]	Any mapping of header values.

Returns

Type	Description
dict[str, str]	Normalised ``dict[str, str]``.

Example

parse_headers({"Content-Type": " application/json "})

{'content-type': 'application/json'}

`parse_url_parts`

parse_url_parts

def parse_url_parts(url: str) -> dict[str, Any]

source

Parse a URL into its component parts.

Parameters

Parameter	Type	Description
`url`	str	Full URL to parse.

Returns

Type	Description
dict[str, Any]	Dictionary with keys ``scheme,` `host,` `port,` `path,` `params,` `fragment``.

`validate_host`

validate_host

def validate_host(host: str) -> None

source

Validate a hostname or IPv4 address.

Parameters

Parameter	Type	Description
`host`	str	Hostname or IP address to validate.

Raises

Exception	Description
HTTPError	If the host is empty or has an invalid format.

`validate_port`

validate_port

def validate_port(port: int) -> None

source

Validate a TCP port number.

Parameters

Parameter	Type	Description
`port`	int	Port number (must be 1–65535).

Raises

Exception	Description
HTTPError	If the port is not an integer or is out of range.

`validate_positive_int`

validate_positive_int

def validate_positive_int(
    value: int,
    field: str = 'value'
) -> None

source

Validate that *value* is a positive integer.

Parameters

Parameter	Type	Description
`value`	int	Integer to validate.
`field`	str	Field name used in the error message.

Raises

Exception	Description
HTTPError	If ``value`` is not a positive integer.

`validate_public_url`

validate_public_url

def validate_public_url(
    url: str,
    *,
    resolve_dns: bool = True,
    fail_on_dns_error: bool = True
) -> None

source

Validate that ``url`` is an ``http``/``https`` URL targeting a public host.

Protects outbound HTTP fetchers (RSS, webhooks, OpenGraph scrapers) against Server-Side Request Forgery (SSRF) targeting loopback, RFC1918/ULA private ranges, cloud metadata link-local addresses (169.254.169.254), and internal DNS suffixes.

Parameters

Parameter	Type	Description
`url`	str	Candidate outbound URL.
`resolve_dns`	bool	Resolve hostnames via ``socket.getaddrinfo`` and verify all returned addresses are globally routable.
`fail_on_dns_error`	bool	When ``True` `(default), raise` `HTTPClientError` `if DNS lookup fails; when` `False``, still validate any resolved IPs while allowing mocked/offline transport layers to handle resolution.

Raises

Exception	Description
HTTPClientError	When ``url`` is malformed, uses a non-HTTP(S) scheme, contains embedded credentials, or targets a non-public host/IP.

`validate_timeout`

validate_timeout

def validate_timeout(timeout: float | None) -> None

source

Validate a timeout value.

Parameters

Parameter	Type	Description
`timeout`	float \| None	Timeout in seconds, or ``None`` for no timeout.

Raises

Exception	Description
HTTPError	If ``timeout`` is not a positive number.

`validate_url`

validate_url

def validate_url(
    url: str,
    require_scheme: bool = True
) -> None

source

Validate a URL string.

Parameters

Parameter	Type	Description
`url`	str	URL to validate.
`require_scheme`	bool	When ``True,` `http/https`` scheme is required.

Raises

Exception	Description
HTTPError	If the URL is missing, malformed, or missing a required scheme.

Exceptions

`HTTPCircuitOpenError`

source

Raised when the circuit breaker is open and requests are rejected.

`HTTPClientError`

source

Base exception for all HTTP client errors.

`HTTPConnectionError`

source

Raised when a connection to a remote host fails.

`HTTPInterceptorError`

source

Raised when a request/response interceptor fails.

`HTTPRetryExhaustedError`

source

Raised when all retry attempts for a request have been exhausted.

`HTTPTimeoutError`

source

Raised when a connection or request times out.