baldur.adapters.fastapi — FastAPI Adapter

fastapi_lifespan async context manager + BaldurMiddleware ASGI middleware. The adapter intentionally stays thin — decisions live in baldur.api.middleware; this wrapper only translates between FastAPI's Request / Response and Baldur's RequestContext / ResponseContext.

The adapter manages no host-app authentication — your app owns endpoint auth, and Baldur's own admin server is gated by key-based roles. Unlike the Django adapter, it registers no Django permission groups.

fastapi

FastAPI adapter for Baldur.

Provides: - init_fastapi(app, service_name=None) — app-construction-time hook. Calls baldur.init() exactly once and installs BaldurMiddleware via app.add_middleware. Mirrors Flask's init_flask; the auto-wiring path because an ASGI lifespan provably cannot add middleware after the stack is built. - fastapi_lifespan — async context manager for FastAPI's lifespan= parameter. Calls baldur.init() on startup; admin server auto-starts via the existing BALDUR_ADMIN_AUTOSTART flag in bootstrap.py. - BaldurMiddleware — ASGI middleware that composes the framework-free helpers from baldur.api.middleware (rate limit, admission / backpressure, CB pre-flight + observation) into a single mountable middleware.

Install: pip install baldur-framework[fastapi]

Example

.. code-block:: python

from fastapi import FastAPI
from baldur.adapters.fastapi import fastapi_lifespan, init_fastapi

app = FastAPI(lifespan=fastapi_lifespan)
init_fastapi(app)

The adapter intentionally stays thin — every decision lives in baldur.api.middleware. The wrapper only translates between FastAPI's Request / Response and Baldur's RequestContext / ResponseContext.

Status: Public

BaldurMiddleware

BaldurMiddleware(
    app: Any,
    service_name: str | None = None,
    rate_limit: int | None = None,
    window_seconds: int | None = None,
)

ASGI middleware integrating Baldur's middleware helpers.

Compatible with FastAPI's app.add_middleware(BaldurMiddleware, ...) and Starlette's middleware stack. Pure ASGI — no FastAPI-specific imports beyond the type hint, so the same class drops into any Starlette-based app.

Exception handling: this middleware does not convert exceptions from the downstream app — it re-raises so Starlette's ServerErrorMiddleware (installed by default, outermost in the stack) still catches unhandled exceptions and emits the 500 response. If the downstream raises before http.response.start is sent, this middleware skips the CB observation (the error never reached the HTTP surface this middleware observes) but records a RED 500 for the latency histogram before re-raising — the metrics analog of Django's _record_exception. A ClientDisconnect is re-raised without a RED record (client fault, not a server 5xx), and BaseException (e.g. asyncio.CancelledError) propagates untouched.

init_fastapi

init_fastapi(
    app: FastAPI,
    service_name: str | None = None,
    rate_limit: int | None = None,
    window_seconds: int | None = None,
) -> None

Initialize Baldur for a FastAPI app.

Parameters:

Name	Type	Description	Default
`app`	`FastAPI`	The FastAPI application instance (must be created before this call — middleware cannot be added once the ASGI stack is built).	required
`service_name`	`str \| None`	Optional upstream identity. When supplied, CB pre-flight + post-response observation are enabled. When `None`, the CB hooks are no-ops (rate limit + admission / backpressure still apply).	`None`
`rate_limit`	`int \| None`	Per-instance override for the middleware rate limit (requests per window). `None` falls back to `RateLimitSettings.middleware_rate_limit` (default `0` = disabled). Pass a positive integer to enable rate limiting only for this FastAPI app.	`None`
`window_seconds`	`int \| None`	Per-instance window size override. `None` falls back to `RateLimitSettings.middleware_window_seconds`.	`None`

fastapi_lifespan `async`

fastapi_lifespan(
    app: FastAPI,
) -> AsyncIterator[dict[str, Any]]

Initialize Baldur on app startup, drain on shutdown.

Yields an empty mapping so callers can attach lifespan-scoped state via the standard lifespan_state pattern; the dict is reserved for Baldur-internal use and may carry diagnostic state in future versions.