API Reference

Protocols

HTTPClientProtocol

source

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.

---

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.

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
) -> None

source

close

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

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.

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.

get

async def get(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a GET request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

post

async def post(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a POST request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

put

async def put(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a PUT request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

delete

async def delete(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a DELETE request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

patch

async def patch(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a PATCH request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

head

async def head(
    url: str,
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Perform a HEAD request.

Parameters

Parameter	Type	Description
`url`	str	Target URL. **kwargs: Forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(HTTPStatusError)`` on 4xx/5xx; ``Err(HTTPConnectionError | HTTPTimeoutError)`` on transport failure.

Raises

Exception	Description
HTTPCircuitOpenError	When the circuit breaker is open.
HTTPRetryExhaustedError	When all retry attempts are exhausted.

post_multipart

async def post_multipart(
    url: str,
    fields: dict[str, str | bytes | tuple[str, bytes, str]],
    **kwargs: Any
) -> Result[HttpResponse, HTTPClientError]

source

Upload multipart/form-data.

Builds an aiohttp.FormData payload from fields and POSTs it. Each value can be:

• str — sent as a plain text field
• bytes — sent as a file-like binary field (filename inferred from the field name)
• (filename, data, content_type) — full control over the part headers

Parameters

Parameter	Type	Description
`url`	str	Target URL.
`fields`	dict[str, str | bytes | tuple[str, bytes, str]]	Mapping of field names to values. See above for supported value types. **kwargs: Additional keyword arguments forwarded to request.

Returns

Type	Description
Result[HttpResponse, HTTPClientError]	``Ok(HttpResponse)`` on 2xx; ``Err(...)`` on failure.

Example

result = await client.post_multipart(
    "https://api.example.com/upload",
    fields={
        "description": "My file",
        "file": ("report.pdf", pdf_bytes, "application/pdf"),
    },
)

result = await client.post_multipart(
    "https://api.example.com/upload",
    fields={
        "description": "My file",
        "file": ("report.pdf", pdf_bytes, "application/pdf"),
    },
)

stream

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

source

Async context manager that streams the response body as raw bytes chunks.

Unlike request, this method does not buffer the entire response body in memory — suitable for large file downloads or chunked responses.

Parameters

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

Yields

Type	Description
AsyncIterator[AsyncIterator[bytes]]	An AsyncIterator of ``bytes`` chunks.

Raises

Exception	Description
HTTPConnectionError	When the connection cannot be established.
HTTPTimeoutError	When the request times out.

Example

async with client.stream("GET", large_file_url) as chunks:
    async for chunk in chunks:
        await file.write(chunk)

async with client.stream("GET", large_file_url) as chunks:
    async for chunk in chunks:
        await file.write(chunk)

sse

async def sse(
    url: str,
    **kwargs: Any
) -> AsyncIterator[AsyncIterator[ServerSentEvent]]

source

Async context manager that consumes a Server-Sent Events (SSE) stream.

Opens a persistent GET connection and parses the text/event-stream response into ServerSentEvent objects.

Parameters

Parameter	Type	Description
`url`	str	SSE endpoint URL. **kwargs: Additional keyword arguments forwarded to ``aiohttp``.

Yields

Type	Description
AsyncIterator[AsyncIterator[ServerSentEvent]]	An AsyncIterator of ServerSentEvent objects.

Raises

Exception	Description
HTTPConnectionError	When the connection cannot be established.

Example

from lexigram.logging import get_logger

logger = get_logger(__name__)
async with client.sse("https://api.example.com/events") as events:
    async for event in events:
        logger.info("sse_event", event_type=event.event, data=event.data)

from lexigram.logging import get_logger

logger = get_logger(__name__)
async with client.sse("https://api.example.com/events") as events:
    async for event in events:
        logger.info("sse_event", event_type=event.event, data=event.data)

session_context

async def session_context(
    cls,
    config: HTTPClientConfig | None = None,
    retry_policy: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None,
    interceptors: Iterable[InterceptorProtocol] = ()
) -> AsyncIterator[HTTPClient]

source

Async context manager that starts and stops the client automatically.

Parameters

Parameter	Type	Description
`config`	HTTPClientConfig | None	Optional HTTPClientConfig; framework defaults apply.
`retry_policy`	RetryPolicyProtocol | None	Optional retry policy; framework default when ``None``.
`circuit_breaker`	CircuitBreakerProtocol | None	Optional circuit breaker; disabled when ``None``.
`interceptors`	Iterable[InterceptorProtocol]	Zero or more interceptors.

Yields

Type	Description
AsyncIterator[HTTPClient]	A started HTTPClient instance.

Example

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

---

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 all requests made by the client. trust_env: Whether to read proxy settings from environment variables (HTTP_PROXY, HTTPS_PROXY, NO_PROXY). Defaults to True — set to False to ignore environment proxies. cookie_jar: Whether to enable an in-memory cookie jar that persists cookies across requests within a single HTTPClient session. Defaults to True.

---

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 lexigram.http.config import HTTPClientConfig

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

from lexigram.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 | None = None) -> None

source

from_config

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

source

Create provider from config object.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register HTTPClient bindings with the container.

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.

---

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_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.

---