Httpware

https://github.com/modern-python/httpware
Admin
Httpware is a Python HTTP client framework with sync and async clients for building resilient...
Tokens:35,516
Snippets:281
Trust Score:9
License:MIT
Update:1 month ago
Show doc for...
Tokens:
Raw
Tokens:
Raw
### Quick start: clone and run just install, lint, test

Source: https://github.com/modern-python/httpware/blob/main/docs/dev/contributing.md

Clone the repository and run the standard setup, lint, and test commands using `just`. The `just install` command runs `uv lock --upgrade` and `uv sync --all-extras --frozen --group lint`; `just lint` runs `eof-fixer`, `ruff format`, `ruff check`, and `ty check`; `just test` runs `pytest` with coverage.

```bash
git clone https://github.com/modern-python/httpware.git
cd httpware
just install        # uv lock --upgrade && uv sync --all-extras --frozen --group lint
just lint           # eof-fixer + ruff format + ruff check + ty check
just test           # pytest with coverage
```

--------------------------------

### Install httpware[otel] extra

Source: https://github.com/modern-python/httpware/blob/main/docs/observability.md

Install the OpenTelemetry extra to enable span event emission. This only installs opentelemetry-api; an SDK and instrumentor are needed separately.

```bash
pip install httpware[otel]
```

--------------------------------

### Install httpware

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Installs the httpware package from PyPI.

```bash
pip install httpware
```

--------------------------------

### Minimal OpenTelemetry SDK and HTTPX instrumentor setup

Source: https://github.com/modern-python/httpware/blob/main/docs/observability.md

Minimal OpenTelemetry setup with console exporter and HTTPX instrumentor. After this runs, every httpware HTTP call gets an HTTP <method> span, and AsyncRetry/AsyncBulkhead events appear as span events on it. For production, replace ConsoleSpanExporter with an OTLP/Jaeger/Zipkin exporter.

```python
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
HTTPXClientInstrumentor().instrument()
```

--------------------------------

### Typed GET request with AsyncClient and pydantic

Source: https://github.com/modern-python/httpware/blob/main/README.md

Quickstart example showing a typed GET request using AsyncClient and a pydantic BaseModel. Requires 'pip install httpware[pydantic]'. The response body is decoded to the User model; a 4xx/5xx raises a typed StatusError, and a malformed body raises DecodeError.

```python
import asyncio

from httpware import AsyncClient
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str


async def main() -> None:
    async with AsyncClient(base_url="https://jsonplaceholder.typicode.com") as client:
        user = await client.get("/users/1", response_model=User)
        print(user.name)  # Leanne Graham


asyncio.run(main())
```

--------------------------------

### Install httpware with optional dependencies

Source: https://github.com/modern-python/httpware/blob/main/README.md

Installation commands for httpware core and optional decoders (pydantic, msgspec) and OpenTelemetry. Use the appropriate extra to enable the desired decoder or all features.

```bash
pip install httpware                # core only — no decoder
pip install httpware[pydantic]      # + PydanticDecoder — BaseModel, dataclasses, primitives, generics
pip install httpware[msgspec]       # + MsgspecDecoder — Struct, dataclasses, primitives, generics
pip install httpware[pydantic,msgspec]   # both — BaseModel routes to pydantic, Struct to msgspec
pip install httpware[all]           # everything (pydantic, msgspec, otel)
```

--------------------------------

### Sync GET request with Client

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Shows the sync usage of Client for a simple GET request. Uses a context manager to manage the client lifecycle. The response is printed as JSON.

```python
from httpware import Client

with Client(base_url="https://jsonplaceholder.typicode.com") as client:
    response = client.get("/users/1")
    print(response.json())
```

--------------------------------

### Compose resilience middleware with AsyncClient

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Complete example of composing resilience middleware in the recommended order with AsyncClient.

```python
from httpware import AsyncClient
from httpware.middleware.resilience import (
    AsyncBulkhead,
    AsyncCircuitBreaker,
    AsyncRetry,
    AsyncTimeout,
)


async def main() -> None:
    async with AsyncClient(
        base_url="https://api.example.com",
        middleware=[
            AsyncTimeout(timeout=30.0),
            AsyncCircuitBreaker(),
            AsyncBulkhead(max_concurrent=10),
            AsyncRetry(),
        ],
    ) as client:
        await client.get("/users/1")
```

--------------------------------

### Async GET request with AsyncClient

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Shows the async usage of AsyncClient for a simple GET request. Requires Python 3.7+ and the httpware package. The response is printed as JSON.

```python
import asyncio

from httpware import AsyncClient


async def main() -> None:
    async with AsyncClient(base_url="https://jsonplaceholder.typicode.com") as client:
        response = await client.get("/users/1")
        print(response.json())


asyncio.run(main())
```

--------------------------------

### Install httpware with optional decoding extras

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Installs optional extras for typed response decoding. The pydantic extra enables PydanticDecoder for BaseModel, dataclasses, primitives, and generics; the msgspec extra enables MsgspecDecoder for Struct, dataclasses, primitives, and generics. Installing both registers both decoders, routing BaseModel to pydantic and Struct to msgspec.

```bash
pip install httpware[pydantic]   # PydanticDecoder — handles BaseModel + dataclasses + primitives + generics
```

```bash
pip install httpware[msgspec]    # MsgspecDecoder — handles Struct + dataclasses + primitives + generics
```

```bash
pip install httpware[pydantic,msgspec]   # both extras — both decoders register; BaseModel routes to pydantic, Struct to msgspec
```

--------------------------------

### AsyncKeyedCircuitBreaker / KeyedCircuitBreaker

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Documentation for the AsyncKeyedCircuitBreaker and KeyedCircuitBreaker classes, including constructor parameters, circuit lifetime, observability, and usage examples.

```APIDOC
## AsyncKeyedCircuitBreaker / KeyedCircuitBreaker

### Description
Creates a circuit breaker that maintains a separate circuit for each circuit key, allowing independent failure handling for multiple upstreams. The async version uses `AsyncKeyedCircuitBreaker` and the sync version uses `KeyedCircuitBreaker`.

### Constructor

Every `AsyncCircuitBreaker` parameter, with the same defaults, plus:

| Parameter | Default | Effect |
|---|---|---|
| `key` | the request's origin | Maps a request to its circuit key. Any hashable value works. The default is the origin as a string built from scheme, host and port, such as `https://a.example` or `http://a.example:8080`. Hosts are lowercased and default ports dropped, so `https://A.example/x` and `https://a.example:443/y` share a circuit, while `http://a.example` and `https://a.example:8443` each get their own. Userinfo is never part of it. |

### Circuit lifetime

A circuit is created on the first request for its key and kept for as long as the breaker lives. Nothing is evicted or pruned, so memory grows with the number of distinct keys. `key` must therefore map requests to a small, bounded set, such as your configured upstreams. Never derive it from user input. A key function that returns something unbounded, such as the full URL, grows the map forever.

### Observability

The keyed breakers emit the same events as `AsyncCircuitBreaker` on the same `httpware.circuit_breaker` logger. Each event has one extra attribute, `circuit_key`, which is the `str()` of the request's circuit key. httpware redacts `url` but not `circuit_key`, which reaches log records and span events exactly as the key function returned it. A custom `key` must not return credentials, tokens or other sensitive values. The default origin contains none.

### Usage

```python
from httpware import AsyncClient
from httpware.middleware.resilience import AsyncKeyedCircuitBreaker, AsyncRetry

async with AsyncClient(
    middleware=[AsyncKeyedCircuitBreaker(failure_threshold=5, reset_timeout=60.0), AsyncRetry()],
) as client:
    await client.get("https://suggest-a.example/v1/suggest")
    await client.get("https://suggest-b.example/v1/suggest")  # unaffected if suggest-a is down
```

Sync usage is identical: `Client` + `KeyedCircuitBreaker`, no `await`.
```

--------------------------------

### RequestIdMiddleware for request-ID propagation

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

Worked example of a RequestIdMiddleware that assigns a per-call UUID, injects it as an outgoing header, and logs it alongside the response status. Place it OUTSIDE AsyncRetry so all attempts of the same call share one ID. The example logs under myapp.request_id, not under httpware.*, which is reserved for library events.

```python
import logging
import uuid

import httpx2

from httpware import AsyncClient, AsyncRetry
from httpware import AsyncNext


_LOGGER = logging.getLogger("myapp.request_id")


class RequestIdMiddleware:
    """Assign a per-call X-Request-Id; log it on response.

    Place OUTSIDE AsyncRetry so all attempts of the same call share one ID
    (so a single call's retries all surface under the same correlation
    key in your logs, and match the URL attribute on httpware.retry's
    emitted events).
    """

    def __init__(self, *, header: str = "X-Request-Id") -> None:
        self._header = header

    async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response:  # noqa: A002
        request_id = str(uuid.uuid4())
        request.headers[self._header] = request_id
        response = await next(request)
        _LOGGER.info(
            "request complete",
            extra={"request_id": request_id, "status": response.status_code},
        )
        return response


async def main() -> None:
    async with AsyncClient(
        base_url="https://api.example.com",
        middleware=[RequestIdMiddleware(), AsyncRetry()],  # ID outside AsyncRetry
    ) as client:
        await client.get("/users/1")
```

--------------------------------

### Pagination loop with send_with_response

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/link-header-pagination.md

Walks all pages of a paginated API by following the Link header. Uses send_with_response to get both the decoded list of Tag objects and the response headers, then extracts the next URL from the 'link' header. The caller must define process and next_link; params is set to None after the first request because the next link carries the query.

```python
from httpware import AsyncClient
from pydantic import BaseModel


class Tag(BaseModel):
    name: str


async def main() -> None:
    async with AsyncClient(base_url="https://gitlab.example/api/v4") as client:
        url = "/projects/1/repository/tags"
        params: dict[str, str] | None = {"per_page": "100", "page": "1"}
        while url:
            request = client.build_request("GET", url, params=params)
            response, tags = await client.send_with_response(request, response_model=list[Tag])
            for tag in tags:
                process(tag)
            url = next_link(response.headers.get("link"))  # caller's parser
            params = None  # next link carries query
```

--------------------------------

### Share RetryBudget across AsyncClients

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Shows how to share a single RetryBudget instance across multiple AsyncClient instances that hit the same downstream, creating one joint budget. The example creates two clients with the same budget and runs concurrent requests with asyncio.gather.

```python
import asyncio

from httpware import AsyncClient
from httpware.middleware.resilience import AsyncRetry, RetryBudget


shared = RetryBudget()


async def main() -> None:
    async with (
        AsyncClient(base_url="https://api.example.com", middleware=[AsyncRetry(budget=shared)]) as users,
        AsyncClient(base_url="https://api.example.com", middleware=[AsyncRetry(budget=shared)]) as orders,
    ):
        await asyncio.gather(users.get("/users/1"), orders.get("/orders/1"))
```

--------------------------------

### AsyncCircuitBreaker / CircuitBreaker

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Documentation for the AsyncCircuitBreaker and CircuitBreaker classes, including constructor parameters and usage examples.

```APIDOC
## AsyncCircuitBreaker / CircuitBreaker

### Description
Creates a circuit breaker that tracks failures and opens the circuit when the failure threshold is reached, preventing further requests until the reset timeout expires. The async version uses `AsyncCircuitBreaker` and the sync version uses `CircuitBreaker`.

### Constructor

| Parameter | Default | Description |
|---|---|---|
| `failure_threshold` | (required) | Number of consecutive failures that will open the circuit. |
| `reset_timeout` | (required) | Time in seconds after which the circuit will attempt to close again. |

### Usage

```python
from httpware import AsyncClient
from httpware.middleware.resilience import AsyncCircuitBreaker

breaker = AsyncCircuitBreaker(failure_threshold=3, reset_timeout=60.0)

async with AsyncClient(
    base_url="https://api.example.com",
    middleware=[breaker],
) as client:
    response = await client.get("/users/1")
```

Sync usage is identical: `Client` + `CircuitBreaker`, no `await`.

### Sharing

Pass the same instance to multiple clients to enforce one shared circuit across them. A `CircuitBreaker` (sync) cannot be shared with an `AsyncCircuitBreaker` — they use different concurrency primitives.
```

--------------------------------

### Compose AsyncBulkhead and AsyncRetry middleware

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Shows composing resilience middleware: AsyncBulkhead outside AsyncRetry so one slot covers all retry attempts. AsyncBulkhead caps total in-flight requests; AsyncRetry uses default 3 attempts with full-jitter backoff. Requires the User model from the previous example.

```python
from httpware import AsyncClient, AsyncBulkhead, AsyncRetry


async def main() -> None:
    async with AsyncClient(
        base_url="https://api.example.com",
        middleware=[
            AsyncBulkhead(max_concurrent=10),  # cap total in-flight
            AsyncRetry(),  # default: 3 attempts, full-jitter backoff
        ],
    ) as client:
        user = await client.get("/users/1", response_model=User)
```

--------------------------------

### Implement a CSV decoder for httpware

Source: https://github.com/modern-python/httpware/blob/main/docs/decoders.md

A custom decoder for text/csv endpoints that returns a list of dataclass rows. The CsvDecoder claims only list[<dataclass>] models and coerces each CSV cell with its field's type. Wire it ahead of the built-in PydanticDecoder so it gets first refusal on list[...] models; the same instance works with a sync Client.

```python
import csv
import dataclasses
import io
import typing

from httpware import AsyncClient
from httpware.decoders.pydantic import PydanticDecoder

T = typing.TypeVar("T")


class CsvDecoder:
    """Decode a text/csv body into a list of dataclass rows.

    Claims only `list[<dataclass>]`; declines everything else so the JSON
    decoders keep their models.
    """

    def can_decode(self, model: type) -> bool:
        if typing.get_origin(model) is not list:
            return False
        args = typing.get_args(model)
        return len(args) == 1 and dataclasses.is_dataclass(args[0])

    def decode(self, content: bytes, model: type[T]) -> T:
        (row_type,) = typing.get_args(model)
        field_types = {f.name: f.type for f in dataclasses.fields(row_type)}
        reader = csv.DictReader(io.StringIO(content.decode("utf-8")))
        return [row_type(**{name: field_types[name](value) for name, value in row.items()}) for row in reader]
```

```python
@dataclasses.dataclass
class Sale:
    id: int
    amount: float
    region: str


async def main() -> None:
    async with AsyncClient(
        base_url="https://reports.example.com",
        decoders=[CsvDecoder(), PydanticDecoder()],
    ) as client:
        sales = await client.send(
            client.build_request("GET", "/sales.csv"),
            response_model=list[Sale],
        )
        # sales: list[Sale]
```

--------------------------------

### Mount full-stack demo with multi-phase incident and composed chain

Source: https://github.com/modern-python/httpware/blob/main/docs/demos/full-stack.md

Mounts the full-stack demo with a multi-phase incident scenario. The fault function defines three phases: latency spike, brownout, and hard down. The chainB configuration shows the recommended composition order: timeout, circuit breaker, bulkhead, retry, and budget. The buildStops array adds explanatory callouts at specific times, including a note that the breaker trips outside the retry loop.

```javascript
document.addEventListener('DOMContentLoaded', function () {
  HttpwareDemo.mount('#fs-demo', {
  scenarios: [
    { id: 'incident', label: 'Multi-phase incident', dur: 16.0,
      fault: (now, rnd) => {
        if (now >= 2.0 && now < 5.0) return { ok: true, ms: 4.0, label: 'latency spike' };
        if (now >= 5.0 && now < 8.0) return rnd() < 0.6
          ? { ok: false, ms: 0.6, label: 'brownout' } : { ok: true, ms: 0.6 };
        if (now >= 8.0 && now < 12.0) return { ok: false, ms: 0.3, label: 'hard down' };
        return { ok: true, ms: 0.05 };
      },
      chainB: { timeout: { timeout: 2.0 },
                circuitBreaker: { failureThreshold: 5, resetTimeout: 2.0, successThreshold: 1 },
                bulkhead: { maxConcurrent: 10, acquireTimeout: 0 },
                retry: { maxAttempts: 3, baseDelay: 0.1, maxDelay: 5.0 },
                budget: { ttl: 10.0, minRetriesPerSec: 10.0, percentCanRetry: 0.2 } } },
  ],
  macroStrip: true,
  stageLabel: (now) => now < 2 ? 'healthy'
    : now < 5 ? 'phase 1 — latency spike'
    : now < 8 ? 'phase 2 — brownout'
    : now < 12 ? 'phase 3 — hard down' : 'recovered',
  buildStops: () => [
    { when: (s) => s.now >= 2.4, spot: ['poolB', 'elapsedB'], title: 'Phase 1 — latency spike',
      body: 'Bulkhead caps concurrency so the slow phase can’t exhaust the client; timeout bounds each operation at 2s.' },
    { when: (s) => s.now >= 5.4, spot: ['ifB'], title: 'Phase 2 — brownout',
      body: 'Retry (within budget) recovers many of the transient errors; the budget keeps it from amplifying.' },
    { when: (s) => s.mw.cb && s.mw.cb.state === 'OPEN', spot: ['brkB'], title: 'Phase 3 — hard down',
      body: 'Consecutive failures trip the breaker OUTSIDE the retry loop, so it short-circuits the whole retry sequence — one outcome per exhausted sequence, not per attempt.' },
    { when: (s) => s.now >= 13.5, spot: ['ifA', 'latA', 'ifB', 'latB'], title: 'The whole stack vs nothing',
      body: 'Plain client: cascading meltdown across every phase. httpware: each layer absorbs the phase it’s built for. That is why they compose.' },
  ],
  });
});
```

--------------------------------

### Recommended middleware ordering

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Recommended ordering of resilience middleware. AsyncTimeout outermost, then circuit breaker, bulkhead, retry, and terminal.

```text
AsyncTimeout → AsyncCircuitBreaker → AsyncBulkhead → AsyncRetry → terminal
```

--------------------------------

### Import AsyncKeyedCircuitBreaker and KeyedCircuitBreaker

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import statements for the keyed circuit breakers. AsyncKeyedCircuitBreaker is for async clients, KeyedCircuitBreaker for sync clients.

```python
from httpware.middleware.resilience import AsyncKeyedCircuitBreaker  # async
from httpware.middleware.resilience import KeyedCircuitBreaker  # sync
```

--------------------------------

### Import AsyncRetry

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import the AsyncRetry middleware class from httpware.middleware.resilience.

```python
from httpware.middleware.resilience import AsyncRetry
```

--------------------------------

### Composition

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Documentation for the recommended composition and ordering of resilience middleware.

```APIDOC
## Composition

### Description
Recommended ordering of resilience middleware: `AsyncTimeout → AsyncCircuitBreaker → AsyncBulkhead → AsyncRetry → terminal`.

### Ordering Rationale
- `AsyncTimeout` outermost so the overall deadline covers the entire sequence including retries and backoff.
- `AsyncCircuitBreaker` outside `AsyncRetry` so an open circuit short-circuits the whole retry loop without attempting any calls. This also means the breaker counts one outcome per fully-exhausted retry sequence rather than one per individual attempt. Placing it outside `AsyncBulkhead` too means a request the open circuit rejects never consumes a concurrency slot.
- `AsyncBulkhead` outside `AsyncRetry` so one slot covers all retry attempts of a single call. Flip those two (`[AsyncRetry, AsyncBulkhead]`) and each retry grabs a fresh slot — defeating the bulkhead under load.

### Example
```python
from httpware import AsyncClient
from httpware.middleware.resilience import (
    AsyncBulkhead,
    AsyncCircuitBreaker,
    AsyncRetry,
    AsyncTimeout,
)


async def main() -> None:
    async with AsyncClient(
        base_url="https://api.example.com",
        middleware=[
            AsyncTimeout(timeout=30.0),
            AsyncCircuitBreaker(),
            AsyncBulkhead(max_concurrent=10),
            AsyncRetry(),
        ],
    ) as client:
        await client.get("/users/1")
```

### Additional Guidance
Cross-cutting middleware that emit per-call state (e.g., the Request-ID middleware in the Middleware guide) should sit outside `AsyncRetry` for the same reason — so all attempts of one call share one ID rather than getting a fresh ID per attempt.
```

--------------------------------

### Import sync Bulkhead middleware

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import the sync Bulkhead middleware for use with Client. It mirrors AsyncBulkhead on a threading.Semaphore and releases the slot via the same try/finally contract for success, exception, and interrupt-style exceptions. Bulkhead and AsyncBulkhead are separate primitives; one instance cannot cap sync and async clients jointly.

```python
from httpware.middleware.resilience import Bulkhead
```

--------------------------------

### Sync Client test with httpx2.MockTransport

Source: https://github.com/modern-python/httpware/blob/main/docs/testing.md

Sync Client test pattern: pass an httpx2.Client (not httpx2.AsyncClient) built on httpx2.MockTransport to Client via httpx2_client=. The handler is defined inline and the response includes request=request.

```python
from http import HTTPStatus

import httpx2

from httpware import Client


def test_get_returns_typed_response() -> None:
    def handler(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(HTTPStatus.OK, request=request, json={"ok": True})

    with Client(httpx2_client=httpx2.Client(transport=httpx2.MockTransport(handler))) as client:
        response = client.get("https://example.test/x")

    assert response.status_code == HTTPStatus.OK
    assert response.json() == {"ok": True}
```

--------------------------------

### Two-step vs one-call shorthand for with_response

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/link-header-pagination.md

Compares the two-step build_request + send_with_response pattern with the equivalent one-call get_with_response shorthand. The two-step form is required when you need full control over the Request object; the shorthand is equivalent for simple cases.

```python
# two-step (pre-built request, required when you need full Request control)
request = client.build_request("GET", url, params=params)
response, tags = await client.send_with_response(request, response_model=list[Tag])

# one-call shorthand (equivalent for the simple case)
response, tags = await client.get_with_response(url, params=params, response_model=list[Tag])
```

--------------------------------

### Import AsyncBulkhead

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import statement for the AsyncBulkhead concurrency limiter.

```python
from httpware.middleware.resilience import AsyncBulkhead
```

--------------------------------

### Sync middleware imports

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

Imports for sync middleware: Middleware, Next, before_request, after_response, on_error from httpware.

```python
from httpware import Middleware, Next, before_request, after_response, on_error
```

--------------------------------

### Import phase decorators from httpware

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

Imports the three phase decorators exported from httpware.middleware. These turn a single async function into an AsyncMiddleware for common cases that don't need state-keeping on self or wrapping the full await next(...) call.

```python
from httpware import async_before_request, async_after_response, async_on_error
```

--------------------------------

### AsyncTimeout

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Documentation for the AsyncTimeout middleware, which bounds total wall-clock time across the entire inner pipeline.

```APIDOC
## AsyncTimeout

### Description
Bounds total wall-clock time across the entire inner pipeline. Place it outermost to enforce that the whole operation must finish within `timeout` seconds, even across retries and backoff sleeps. On expiry it raises `httpware.TimeoutError`.

### Parameters
- `timeout` (float) - Required - Overall deadline in seconds. Must be a finite number > 0; a non-finite (inf/nan) or ≤0 value raises `ValueError`.

### Notes
- This is not a per-call timeout. httpx2's connect/read/write/pool timeouts are the right tool for bounding a single outbound call; `AsyncTimeout` doesn't duplicate them.
- No sync `Timeout` exists. Sync Python has no cancellation primitive that can interrupt a blocking httpx2 call mid-flight. For sync per-call bounds, configure `httpx2.Timeout` on the wrapped client or pass `timeout=` per request.

### Observability
- Event: `timeout.exceeded` on logger `httpware.timeout`.
```

--------------------------------

### Configure AsyncClient middleware with modern-di Factory

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/modern-di.md

Use providers.Factory with kwargs to pass middleware to AsyncClient. Each cached singleton owns its own AsyncBulkhead and AsyncRetry state, suitable for different reliability profiles.

```python
from httpware import AsyncClient, AsyncBulkhead, AsyncRetry


class ServiceClients(Group):
    user_api = providers.Factory(
        scope=Scope.APP,
        creator=UserApi,
        kwargs={
            "base_url": "https://users.example.com",
            "middleware": [AsyncBulkhead(max_concurrent=10), AsyncRetry()],
        },
        cache_settings=providers.CacheSettings(finalizer=UserApi.aclose),
    )
```

--------------------------------

### Import sync Retry middleware

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import the sync Retry middleware for use with Client. It takes the same parameters as AsyncRetry and sleeps with time.sleep between attempts. Retry-After, streaming-body refusal, exhaustion behavior, and RetryBudgetExhaustedError semantics are identical to AsyncRetry.

```python
from httpware.middleware.resilience import Retry
```

--------------------------------

### Import AsyncCircuitBreaker and CircuitBreaker

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Imports the async and sync circuit breaker classes from the resilience middleware module.

```python
from httpware.middleware.resilience import AsyncCircuitBreaker  # async
from httpware.middleware.resilience import CircuitBreaker  # sync
```

--------------------------------

### Async test with httpx2.MockTransport

Source: https://github.com/modern-python/httpware/blob/main/docs/testing.md

Basic async test pattern using httpx2.MockTransport with a sync handler. Pass the pre-built httpx2.AsyncClient to AsyncClient via httpx2_client=; the middleware chain runs end-to-end with only the wire mocked. If using pytest-asyncio in auto-mode, no @pytest.mark.asyncio decorator is needed.

```python
from http import HTTPStatus

import httpx2

from httpware import AsyncClient


def handler(request: httpx2.Request) -> httpx2.Response:
    return httpx2.Response(HTTPStatus.OK, json={"id": 1, "name": "Alice"})


async def test_get_user() -> None:
    transport = httpx2.MockTransport(handler)
    async with AsyncClient(httpx2_client=httpx2.AsyncClient(transport=transport)) as client:
        response = await client.get("https://api.example.test/users/1")
    assert response.status_code == HTTPStatus.OK
    assert response.json()["name"] == "Alice"
```

--------------------------------

### Wire AsyncClient into modern-di container with Factory and finalizer

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/modern-di.md

Creates a ServiceClients group with a Factory provider for AsyncClient, scoped to the application lifetime. The CacheSettings finalizer makes the provider a singleton and ensures AsyncClient.aclose is awaited on container teardown. Resolution is synchronous in modern-di 2.x; use await container.close_async() to run the finalizer.

```python
from modern_di import Container, Group, Scope, providers

from httpware import AsyncClient


class ServiceClients(Group):
    api = providers.Factory(
        scope=Scope.APP,
        creator=AsyncClient,
        kwargs={"base_url": "https://api.example.com"},
        cache_settings=providers.CacheSettings(finalizer=AsyncClient.aclose),
    )


async def main() -> None:
    container = Container(scope=Scope.APP, groups=[ServiceClients])
    try:
        client = container.resolve(AsyncClient)
        response = await client.get("/users/1")
        print(response.status_code)
    finally:
        await container.close_async()  # runs the AsyncClient.aclose finalizer
```

--------------------------------

### Import AsyncTimeout

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import statement for the AsyncTimeout middleware class.

```python
from httpware.middleware.resilience import AsyncTimeout
```

--------------------------------

### Catch httpware exceptions in async fetch

Source: https://github.com/modern-python/httpware/blob/main/docs/errors.md

Shows how to catch httpware exceptions in an async client call. Assumes a module logger named `_LOGGER` in your own namespace (not under `httpware.*`). `NotFoundError` is the most precise status exception; `StatusError` exposes `exc.response.*`; `NetworkError` indicates retries were exhausted or the method was non-idempotent; `RetryBudgetExhaustedError` and `BulkheadFullError` signal backpressure; `ClientError` is the catch-all for anything else httpware raised.

```python
import logging

from httpware import (
    AsyncClient,
    ClientError,
    StatusError,
    NetworkError,
    TimeoutError,
    NotFoundError,
    RetryBudgetExhaustedError,
    BulkheadFullError,
)

_LOGGER = logging.getLogger("myapp")


async def fetch(client: AsyncClient, user_id: int) -> dict | None:
    try:
        return await client.get(f"/users/{user_id}", response_model=dict)
    except NotFoundError:
        # Specific status — most precise. Convert to None as the "absent" sentinel.
        return None
    except StatusError as exc:
        # Got a response, but its status was 4xx/5xx and not one we handle specifically.
        # exc.response.* is available — headers, content, request, etc.
        _LOGGER.warning("upstream returned %s for %s", exc.response.status_code, exc.response.request.url)
        raise
    except NetworkError:
        # Transient transport failure. Already retried by the default AsyncRetry middleware
        # (if installed) when the method was idempotent. Seeing this means retries
        # exhausted or the method was non-idempotent.
        raise
    except (RetryBudgetExhaustedError, BulkheadFullError) as exc:
        # Resilience refusal — backpressure signal. Back off the caller.
        _LOGGER.error("resilience refused: %s", exc)
        raise
    except ClientError:
        # Catch-all for anything else httpware raised.
        raise
```

--------------------------------

### Sync logging middleware with Client

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

A sync middleware class that logs requests and responses, used with a sync Client. The middleware is a structural protocol; any callable with the right signature satisfies it.

```python
import logging

import httpx2

from httpware import Client
from httpware import Next


_LOGGER = logging.getLogger("myapp.logging_middleware")


class LoggingMiddleware:
    def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # noqa: A002
        _LOGGER.info("-> %s %s", request.method, request.url)
        response = next(request)
        _LOGGER.info("<- %s", response.status_code)
        return response


with Client(base_url="https://api.example.com", middleware=[LoggingMiddleware()]) as client:
    client.get("/users/1")
```

--------------------------------

### Typed decoding with response_model= in AsyncClient

Source: https://github.com/modern-python/httpware/blob/main/docs/index.md

Demonstrates typed decoding with response_model= in async usage. Requires pydantic. The response is decoded into a User model and the name attribute is printed.

```python
from httpware import AsyncClient
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str


async def main() -> None:
    async with AsyncClient(base_url="https://api.example.com") as client:
        user = await client.get("/users/1", response_model=User)
        print(user.name)
```

--------------------------------

### RetryBudget token-bucket formula

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Formula for the token-bucket ceiling: the percent term rounds up with math.ceil, the floor term truncates with int. A withdrawal fails when len(withdrawn_in_window) >= ceiling.

```text
ceiling = ceil(len(deposits_in_window) * percent_can_retry) + int(min_retries_per_sec * ttl)
```

--------------------------------

### Configure AsyncCircuitBreaker with failure_rate_threshold

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Configures an AsyncCircuitBreaker in time-based failure-rate mode, opening when at least 50% of calls fail over a rolling 30-second window, but only after at least 20 calls are observed. The same parameters apply to the sync CircuitBreaker.

```python
from httpware.middleware.resilience import AsyncCircuitBreaker


breaker = AsyncCircuitBreaker(
    failure_rate_threshold=0.5,  # open at ≥50% failures
    window_seconds=30.0,  # over a rolling 30s window
    minimum_calls=20,  # but only once 20+ calls are observed
)
```

--------------------------------

### AsyncMiddleware protocol and AsyncNext type alias

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

Defines the AsyncMiddleware protocol and AsyncNext type alias. The chain is composed once at AsyncClient.__init__ and frozen for the client's lifetime; the first entry in middleware=[...] is the outermost layer. Calling await next(request) forwards to the next layer, and you must return an httpx2.Response.

```python
from collections.abc import Awaitable, Callable
from typing import Protocol, TypeAlias, runtime_checkable
import httpx2

AsyncNext: TypeAlias = Callable[[httpx2.Request], Awaitable[httpx2.Response]]


@runtime_checkable
class AsyncMiddleware(Protocol):
    async def __call__(self, request: httpx2.Request, next: AsyncNext) -> httpx2.Response: ...
```

--------------------------------

### Import RetryBudget

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Import statement for the RetryBudget class from the httpware resilience middleware module.

```python
from httpware.middleware.resilience import RetryBudget
```

--------------------------------

### Testing custom middleware with mock transport

Source: https://github.com/modern-python/httpware/blob/main/docs/testing.md

Compose your custom middleware with the mock transport to exercise the chain end-to-end. Uses the _ResponseSequence handler to capture the request and assert on the header added by MyHeaderMiddleware. For state-keeping middleware, assert on instance attributes after the call.

```python
async def test_my_middleware_adds_header() -> None:
    handler = _ResponseSequence([HTTPStatus.OK])
    async with AsyncClient(
        httpx2_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)),
        middleware=[MyHeaderMiddleware()],
    ) as client:
        await client.get("https://example.test/x")
    assert handler.calls[0].headers["X-My-Header"] == "expected-value"
```

--------------------------------

### Use @async_after_response to count responses by status class

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/phase-decorator-patterns.md

Defines a factory that returns a middleware incrementing a metric keyed by status class for each response. The decorated function returns the response unchanged. Shows a noop sink and usage inside an AsyncClient with the middleware.

```python
from collections.abc import Callable

import httpx2

from httpware import AsyncClient
from httpware import AsyncMiddleware, async_after_response


MetricSink = Callable[[str, int], None]


def status_class_counter(metric_sink: MetricSink) -> AsyncMiddleware:
    @async_after_response
    async def observe(request: httpx2.Request, response: httpx2.Response) -> httpx2.Response:
        status_class = f"{response.status_code // 100}xx"
        metric_sink(f"http.{request.method.lower()}.responses.{status_class}", 1)
        return response

    return observe


def noop_sink(name: str, count: int) -> None:
    """Replace with your statsd / Prometheus / Datadog client."""


async def main() -> None:
    async with AsyncClient(
        base_url="https://api.example.com",
        middleware=[status_class_counter(noop_sink)],
    ) as client:
        await client.get("/me")
```

--------------------------------

### Wayfinding operations with gh CLI

Source: https://github.com/modern-python/httpware/blob/main/docs/agents/issue-tracker.md

Commands for wayfinding operations: creating the map issue, adding blocking dependencies via the sub-issues endpoint, querying the blocker's database id, listing open issues, claiming a ticket, and resolving it. The blocker-db-id is the numeric database id, not the #number or node_id.

```bash
gh issue create --label wayfinder:map
```

```bash
gh api --method POST repos/<owner>/<repo>/issues/<child>/dependencies/blocked_by -F issue_id=<blocker-db-id>
```

```bash
gh api repos/<owner>/<repo>/issues/<n> --jq .id
```

```bash
gh issue list --state open
```

```bash
gh issue edit <n> --add-assignee @me
```

```bash
gh issue comment <n> --body "<answer>"
```

```bash
gh issue close <n>
```

--------------------------------

### Failing: Duplicate provider type collision with AsyncClient

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/modern-di.md

This snippet shows the problematic configuration where two providers use the same creator AsyncClient, causing a DuplicateProviderTypeError at container construction. The error message indicates the provider is duplicated by type.

```python
class ServiceClients(Group):
    user_api = providers.Factory(
        scope=Scope.APP,
        creator=AsyncClient,
        kwargs={"base_url": "https://users.example.com"},
        cache_settings=providers.CacheSettings(finalizer=AsyncClient.aclose),
    )
    billing_api = providers.Factory(
        scope=Scope.APP,
        creator=AsyncClient,
        kwargs={"base_url": "https://billing.example.com"},
        cache_settings=providers.CacheSettings(finalizer=AsyncClient.aclose),
    )


# At Container(...) construction:
# modern_di.exceptions.DuplicateProviderTypeError: Provider is duplicated by type
# <class 'httpware.client.AsyncClient'>. To resolve this issue: ...
```

--------------------------------

### Issue tracker operations with gh CLI

Source: https://github.com/modern-python/httpware/blob/main/docs/agents/issue-tracker.md

Commands for creating, reading, listing, commenting on, labeling, and closing GitHub issues using the gh CLI. Use a heredoc for multi-line bodies. The repo is inferred from git remote -v automatically when run inside a clone.

```bash
gh issue create --title "..." --body "..."
```

```bash
gh issue view <number> --comments
```

```bash
gh issue list --state open --json number,title,body,labels,comments --jq '[.[] | {number, title, body, labels: [.labels[].name], comments: [.comments[].body]}]'
```

```bash
gh issue comment <number> --body "..."
```

```bash
gh issue edit <number> --add-label "..."
```

```bash
gh issue edit <number> --remove-label "..."
```

```bash
gh issue close <number> --comment "..."
```

--------------------------------

### Enable logging for httpware resilience events

Source: https://github.com/modern-python/httpware/blob/main/docs/observability.md

Configure logging levels for the resilience middleware loggers. Set the circuit breaker logger to INFO to see recovery events; others to WARNING. This is the stable public contract for logger names.

```python
import logging

# Enable visibility into resilience operational events
logging.getLogger("httpware.retry").setLevel(logging.WARNING)
logging.getLogger("httpware.bulkhead").setLevel(logging.WARNING)
logging.getLogger("httpware.circuit_breaker").setLevel(logging.INFO)  # INFO for recovery events
logging.getLogger("httpware.timeout").setLevel(logging.WARNING)
```

--------------------------------

### Fix: Subclass AsyncClient per backend to avoid type collision

Source: https://github.com/modern-python/httpware/blob/main/docs/recipes/modern-di.md

This snippet shows the corrected approach: define typing-only subclasses UserApi and BillingApi of AsyncClient, then use them as creators in the providers. This gives each provider a distinct bound_type, allowing container.resolve() to route correctly. The subclasses have empty bodies and inherit all methods unchanged.

```python
from modern_di import Container, Group, Scope, providers

from httpware import AsyncClient


class UserApi(AsyncClient):
    """Typing handle for the User service backend."""


class BillingApi(AsyncClient):
    """Typing handle for the Billing service backend."""


class ServiceClients(Group):
    user_api = providers.Factory(
        scope=Scope.APP,
        creator=UserApi,
        kwargs={"base_url": "https://users.example.com"},
        cache_settings=providers.CacheSettings(finalizer=UserApi.aclose),
    )
    billing_api = providers.Factory(
        scope=Scope.APP,
        creator=BillingApi,
        kwargs={"base_url": "https://billing.example.com"},
        cache_settings=providers.CacheSettings(finalizer=BillingApi.aclose),
    )


async def main() -> None:
    container = Container(scope=Scope.APP, groups=[ServiceClients])
    try:
        users = container.resolve(UserApi)
        billing = container.resolve(BillingApi)
        # ... use them
    finally:
        await container.close_async()
```

--------------------------------

### Streaming-body refusal note

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

PEP 678 note added when AsyncRetry refuses to retry a request with a streaming body.

```text
httpware: not retrying — request body is a stream that cannot replay across attempts
```

--------------------------------

### Inspect circuit breaker state with CircuitState

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Shows how to inspect the circuit breaker's state using the read-only `state` property and the `CircuitState` enum, typically in a health or readiness handler. The state property is read-only and does not trigger the lazy OPEN to HALF_OPEN transition.

```python
from httpware import CircuitState
from httpware.middleware.resilience import AsyncCircuitBreaker

breaker = AsyncCircuitBreaker(failure_threshold=5)
# ... later, in a health/readiness handler:
if breaker.state is CircuitState.OPEN:
    ...  # report the dependency as degraded
```

--------------------------------

### Stateful handler for retry testing with AsyncRetry

Source: https://github.com/modern-python/httpware/blob/main/docs/testing.md

Stateful handler that returns each status in order and records every request. Used to test AsyncRetry: the tiny base_delay/max_delay make the test run instantly without freezegun or sleep injection. Asserts the retry occurred by checking handler.calls length.

```python
from httpware import AsyncRetry


class _ResponseSequence:
    """Returns each status in order; records every request received."""

    def __init__(self, statuses: list[int]) -> None:
        self._statuses = list(statuses)
        self.calls: list[httpx2.Request] = []

    def __call__(self, request: httpx2.Request) -> httpx2.Response:
        self.calls.append(request)
        status = self._statuses.pop(0) if self._statuses else HTTPStatus.OK
        return httpx2.Response(status, request=request)


async def test_retry_succeeds_after_503() -> None:
    handler = _ResponseSequence([HTTPStatus.SERVICE_UNAVAILABLE, HTTPStatus.OK])
    transport = httpx2.MockTransport(handler)
    async with AsyncClient(
        httpx2_client=httpx2.AsyncClient(transport=transport),
        middleware=[AsyncRetry(base_delay=0.001, max_delay=0.002)],
    ) as client:
        response = await client.get("https://example.test/x")
    assert response.status_code == HTTPStatus.OK
    assert len(handler.calls) == 2  # initial + 1 retry
```

--------------------------------

### Share AsyncBulkhead across clients

Source: https://github.com/modern-python/httpware/blob/main/docs/resilience.md

Shares a single AsyncBulkhead instance across multiple AsyncClient instances to cap combined in-flight requests.

```python
shared_bulkhead = AsyncBulkhead(max_concurrent=10)

async with (
    AsyncClient(base_url="https://api.example.com", middleware=[shared_bulkhead, AsyncRetry()]) as a,
    AsyncClient(base_url="https://api.example.com", middleware=[shared_bulkhead, AsyncRetry()]) as b,
):
    ...  # combined in-flight across a + b is capped at 10
```

--------------------------------

### Sync before_request decorator

Source: https://github.com/modern-python/httpware/blob/main/docs/middleware.md

A sync middleware using the @before_request decorator to add a request ID header. The decorator wraps a sync function and is used with a sync Client.

```python
import uuid

import httpx2

from httpware import Client, before_request


@before_request
def add_request_id(request: httpx2.Request) -> httpx2.Request:
    return httpx2.Request(
        request.method,
        request.url,
        headers={**request.headers, "X-Request-ID": uuid.uuid4().hex},
        content=request.content,
    )


with Client(base_url="https://api.example.com", middleware=[add_request_id]) as client:
    client.get("/users/1")
```