FastAPI

Start here: Request handlers and the ambient scope
Common recipes: micro.install(app) wires the lifespan and binds the active app inside request handlers, so patterns resolve their backends ambiently without explicit backend= wiring. uses=[OpsServer()] serves /livez, /readyz, /healthz, and /metrics on a port of their own, which is the recommended way. health_router() and metrics_router() serve them on the app port instead, where every caller reaches them, so gate /healthz and /metrics there. uses=[IdempotentRequests()] replays a stored response when a request repeats its Idempotency-Key, uses=[ConditionalRequests()] documents If-Match and If-None-Match in the schema so Swagger offers them, and dependencies=[CachedResponse(ttl=60)] answers a repeated read of one route from the cache. principal: CurrentPrincipal reads the caller AuthenticatedRequests verified, and dependencies=[Authenticated(scopes=[...])] requires scopes.

Everything pure ASGI lives in Starlette, App, and HTTP. This module adds what only FastAPI has: the route dependencies and the health router.

grelmicro.integrations.fastapi

FastAPI integration: the Starlette wiring plus what Starlette lacks.

The lifespan, the binding, and the error responses are pure ASGI and live in grelmicro.integrations.starlette. This module adds what Starlette lacks, the OpenAPI schema and the health and metrics routers.

CurrentPrincipal `module-attribute`

CurrentPrincipal = Principal

The authenticated caller, whatever proved who it is.

from grelmicro.integrations.fastapi import CurrentPrincipal


@app.get("/orders")
async def list_orders(principal: CurrentPrincipal) -> list[Order]:
    return await orders_of(principal.issuer, principal.subject)

A request that reached the route without a credential is answered 401. Read inside Security(..., scopes=[...]), it requires those scopes too, and a caller lacking one is answered 403. It needs a registered AuthenticatedRequests, which verifies the token before the route runs.

CurrentToken `module-attribute`

CurrentToken = VerifiedToken

The bearer token the caller presented, once AuthenticatedRequests verified it.

from grelmicro.integrations.fastapi import CurrentToken


@app.post("/orders")
async def create_order(token: CurrentToken) -> dict[str, str]:
    response = await payments.post(
        "/charges", auth=payments_for_user.auth(token)
    )
    return response.json()

Read it to act for the caller, such as exchanging it with a TokenExchange. token.value is the token and token.expires_at when it expires, and its repr never shows the token. A request that reached the route without a credential is answered 401, as CurrentPrincipal answers it.

OptionalPrincipal `module-attribute`

OptionalPrincipal = Principal | None

The caller when the request sent a valid token, and None when it sent none.

from grelmicro.integrations.fastapi import Anonymous, OptionalPrincipal


@app.get("/catalog", dependencies=[Anonymous()])
async def catalog(principal: OptionalPrincipal) -> list[Product]:
    if principal is None:
        return await public_catalog()
    return await catalog_for(principal.subject)

Read on a route declaring Anonymous(), where a credential is optional. A token that does not verify never reaches it: the request is answered 401.

Claims `module-attribute`

Claims = JWTClaims

The verified JWT claims of the caller, typed.

from grelmicro.integrations.fastapi import Claims


@app.get("/me")
async def me(claims: Claims) -> dict[str, str]:
    return {"tenant": claims.claims["tenant"]}

Read CurrentPrincipal instead in a handler that should not care how the caller was authenticated.

Conditional `module-attribute`

Conditional = ConditionalRequest

The conditional guards, injected.

from grelmicro.integrations.fastapi import Conditional


@app.patch("/carts/{cart_id}")
async def update(cart_id: int, conditional: Conditional) -> Cart:
    cart = await load(cart_id)
    conditional.check(cart.version)
    return await save(cart)

A ready-made annotation, so a handler declares one word rather than Annotated[ConditionalRequest, Depends(...)].

ConditionalRequired `module-attribute`

ConditionalRequired = ConditionalRequest

The conditional guards, injected, on a route that requires a precondition.

from grelmicro.integrations.fastapi import ConditionalRequired


@app.put("/carts/{cart_id}")
async def replace(cart_id: int, conditional: ConditionalRequired) -> Cart:
    cart = await load(cart_id)
    conditional.check(cart.version)
    return await save(cart)

A request carrying neither If-Match nor If-None-Match is answered 428 before the handler runs. The route declares precondition_required, so the OpenAPI schema marks If-Match required on its methods, POST included.

ConditionalRequest

ConditionalRequest(
    if_match: str | None = None,
    if_none_match: str | None = None,
)

The conditional guards, as a dependency FastAPI injects.

The same two calls as check_freshness and check_precondition, for a handler that would rather have them injected than imported:

from grelmicro.integrations.fastapi import Conditional


@app.patch("/carts/{cart_id}")
async def update(cart_id: int, conditional: Conditional) -> Cart:
    cart = await load(cart_id)
    conditional.check(cart.version)
    return await save(cart)

It is the same implementation underneath, so a service can use whichever reads better. The plain functions work on Starlette and Litestar too, where there is no dependency injection to hang this on.

check reads the headers this was handed, so a route that injects it is guarded whether or not ConditionalRequests() is registered. fresh needs the component: recording a version is how the middleware knows what tag to put on the response, and without one there is nothing to record it in, so it raises OutOfContextError.

Declaring it puts If-Match and If-None-Match in the schema for that operation alone, so Swagger offers the fields exactly where a handler reads them, whatever the path-based rules say.

Take what the client sent, so a handler can read it raw too.

PARAMETER	DESCRIPTION
`if_match`	What the client sent in `If-Match`, if anything. TYPE: `str \| None` DEFAULT: `None`
`if_none_match`	What the client sent in `If-None-Match`, if anything. TYPE: `str \| None` DEFAULT: `None`

if_match `instance-attribute`

if_match = if_match

if_none_match `instance-attribute`

if_none_match = if_none_match

check

check(
    version: object = _UNSET,
    *,
    etag: str | None = None,
    require: bool = True,
) -> None

Refuse a write whose precondition no longer holds.

PARAMETER	DESCRIPTION
`version`	What identifies the version the resource carries now. TYPE: `object` DEFAULT: `_UNSET`
`etag`	An entity tag that is already one, quotes included. TYPE: `str \| None` DEFAULT: `None`
`require`	Answer `428` when the request carries no precondition. TYPE: `bool` DEFAULT: `True`

RAISES	DESCRIPTION
`PreconditionFailedError`	If the resource is not in the state the request's `If-Match` or `If-None-Match` asked for.
`PreconditionRequiredError`	If `require` and none was sent.

fresh

fresh(
    version: object = _UNSET, *, etag: str | None = None
) -> bool

Answer this read with an entity tag, and say whether it changed.

PARAMETER	DESCRIPTION
`version`	What identifies the version the resource carries now. TYPE: `object` DEFAULT: `_UNSET`
`etag`	An entity tag that is already one, quotes included. TYPE: `str \| None` DEFAULT: `None`

CheckResultResponse

Bases: BaseModel

Health status of a single check.

error is present only on a failing check, and details only when the check returned some and the router is showing them. Both are absent otherwise rather than sent as null, so the schema types them as a plain string and object.

status `instance-attribute`

status: HealthStatus

critical `class-attribute` `instance-attribute`

critical: bool = True

error `instance-attribute`

error: str | SkipJsonSchema[None]

details `instance-attribute`

details: dict[str, Any] | SkipJsonSchema[None]

HealthzResponse

Bases: BaseModel

Aggregate health report.

status `instance-attribute`

status: HealthStatus

checks `instance-attribute`

checks: dict[str, CheckResultResponse]

install

install(
    app: FastAPI, micro: Grelmicro, *, ambient: bool = True
) -> None

Wire micro into a FastAPI app.

The Starlette wiring, plus FastAPI's request telemetry exported through the Trace and Metrics components. The access record and the security event of a request name the route FastAPI's request span names.

Prefer the polymorphic micro.install(app), which detects the framework and calls this for you.

PARAMETER	DESCRIPTION
`app`	The FastAPI application to wire. TYPE: `FastAPI`
`micro`	The `Grelmicro` app to open in the lifespan and bind per request. TYPE: `Grelmicro`
`ambient`	Add `GrelmicroMiddleware` so patterns resolve ambiently inside request handlers. Default `True`. Pass `False` to skip it. TYPE: `bool` DEFAULT: `True`

install_error_responses

install_error_responses(
    app: FastAPI, errors: ErrorResponses
) -> None

Answer every error in the registered format, schema included.

The Starlette handlers, plus the OpenAPI rewriting: FastAPI describes a generated 422 with its own model, which is no longer what those operations answer with.

install_middleware

install_middleware(
    app: FastAPI, components: Sequence[Any]
) -> None

Add each component's ASGI middleware and describe it in the schema.

The Starlette wiring reads route rules and adds the middleware. Each component carrying _document_openapi is handed the schema when FastAPI builds it, with the route declarations and the registered error format, so a route added after this call is described too. A component without it is added silently.

PARAMETER	DESCRIPTION
`app`	The FastAPI application to wire. TYPE: `FastAPI`
`components`	The registered components that carry an ASGI middleware. TYPE: `Sequence[Any]`

is_bound

is_bound(app: Starlette) -> bool

Return whether install added the per-request binding middleware.

Called by Grelmicro.check_ambient_binding and Grelmicro.describe to catch an app that never had micro.install(app) called on it, including a mounted sub-application, which otherwise resolves against the host's components with nothing reporting it.

PARAMETER	DESCRIPTION
`app`	The Starlette or FastAPI application to inspect. TYPE: `Starlette`

error_response

error_response(
    request: Request,
    *,
    status: int,
    detail: str | None = None,
    extensions: dict[str, Any] | None = None,
) -> Response

Answer from your own exception handler in the app's error format.

Writing a handler of your own is how one error opts out of the shape grelmicro installs. This is for the other case: you want your own handler and the same shape as everything else.

from grelmicro.integrations.fastapi import error_response


@app.exception_handler(InsufficientFunds)
async def handle(request: Request, exc: InsufficientFunds) -> Response:
    return error_response(
        request,
        status=409,
        detail="The account does not hold enough to cover this charge.",
        extensions={"balance": exc.balance},
    )

The format is read from the app, so a service that registered ErrorResponses.tmf() answers in TMF from here too, with no second place to keep in step. An app that registered nothing gets RFC 9457.

PARAMETER	DESCRIPTION
`request`	The request being answered, which knows its app. TYPE: `Request`
`status`	HTTP status code of the response. TYPE: `int`
`detail`	Explanation of this occurrence, safe to show a client. TYPE: `str \| None` DEFAULT: `None`
`extensions`	Extra members to carry, where the format has room for them. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

CachedResponse

CachedResponse(
    *,
    ttl: int | timedelta | None = None,
    shared: bool = False,
) -> Any

Declare that this route's response is cached.

Declared on the route rather than called in the handler, because a hit is answered before the handler runs. micro.install(app) reads it off the dependency tree, so a repeated read is answered without the handler running:

from grelmicro.integrations.fastapi import CachedResponse


@app.get("/products", dependencies=[CachedResponse(ttl=60)])
async def list_products() -> list[Product]: ...

It needs a registered CachedResponses(), which holds the store and the rules every route shares. Without one, nothing reads the declaration and the route answers from its handler every time.

@cached caches what a function returns, which is not what the route answers: the framework still has to serialize it, and a header the handler set is not part of it. This caches the response.

Only GET routes take it. micro.install(app) refuses it on a route that answers anything else, naming the path.

Starlette and Litestar resolve no dependencies to hang this on, so they name their paths in CachedResponses(include=...) instead, with how long each is kept for.

RateLimited

RateLimited(
    *limiters: RateLimiter,
    cost: int = 1,
    max_wait: float = 0.0,
    key: Callable[[Any], str | None] | None = None,
    trusted: TrustedProxies | None = None,
    legacy_headers: bool = False,
) -> Any

Meter one route, on top of whatever the app meters.

RateLimitedRequests(...) meters the whole app. Declare this on the route that costs more than the rest, or that has a quota of its own:

from grelmicro.integrations.fastapi import RateLimited


@app.post("/search", dependencies=[RateLimited(searches, cost=5)])
async def search(query: str) -> list[Hit]: ...

Allowed or refused, the response states what the caller has left in the RateLimit and RateLimit-Policy fields, and a refusal answers 429 through the same path every other rejection takes.

A route that returns a Response of its own needs a registered RateLimitedRequests(...) for those fields to reach the wire: a framework merges what a dependency states into the response it builds itself, and not into one a handler already built. The tokens are spent either way.

The caller is the address a middleware already resolved, or the one trusted resolves here. Without either, and without key, the call is left unmetered rather than metered under the ingress.

Authenticated

Authenticated(*, scopes: Sequence[str] = ()) -> Any

Require an authenticated caller holding every scope named.

from grelmicro.integrations.fastapi import Authenticated


@app.delete(
    "/orders/{order_id}",
    dependencies=[Authenticated(scopes=["orders:write"])],
)
async def cancel(order_id: int) -> None: ...

Built on fastapi.Security, so the scopes reach SecurityScopes, and a handler can take the caller from it:

@app.get("/orders")
async def list_orders(
    principal: Annotated[Principal, Authenticated(scopes=["orders:read"])],
) -> list[Order]: ...

A caller with no credential is answered 401, and one lacking a scope 403, whose WWW-Authenticate challenge names the scopes.

It needs a registered AuthenticatedRequests, which verifies the token before the route runs. Without one, every call is answered 401.

Anonymous

Anonymous() -> Any

Serve this route without a credential.

Every route is authenticated once AuthenticatedRequests is registered. Declare this on the one that is public:

from grelmicro.integrations.fastapi import Anonymous


@app.get("/catalog", dependencies=[Anonymous()])
async def catalog() -> list[Product]: ...

micro.install(app) reads it off the dependency tree, and off what the router holding the route was included with. It applies per method, so a path serving a public read keeps its writes authenticated. It is read off the route FastAPI dispatches the request to, so a declaration never opens a route it was not written on. A credential is optional there. A request sending none is served with a caller that is not authenticated, and a bearer token that is sent is verified, so a valid one hands the route its caller through OptionalPrincipal and one that does not verify is answered 401.

health_router

health_router(
    component: HealthChecks | None = None,
    *,
    prefix: str = "",
    include_in_schema: bool = False,
    show_details: bool | Depends = False,
    healthz_dependencies: list[Depends] | None = None,
) -> APIRouter

Create a FastAPI router with health check endpoints.

Provides three endpoints:

GET/HEAD {prefix}/livez: Liveness probe. Runs no check per request. 503 while a liveness check fails, else 200, with an empty body.
GET/HEAD {prefix}/readyz: Readiness probe. Runs critical checkers only. Returns 200 or 503 with an empty body.
GET/HEAD {prefix}/healthz: Aggregate JSON report.

All responses set Cache-Control: no-store. None of them appear in the OpenAPI schema unless include_in_schema=True.

PARAMETER	DESCRIPTION
`component`	Health checks component whose checks the router runs. When omitted, the router resolves the default instance from the active `Grelmicro` app (`Grelmicro(uses=[HealthChecks(...)])`). TYPE: `HealthChecks \| None` DEFAULT: `None`
`prefix`	URL prefix for health endpoints (e.g. '/api/v1'). TYPE: `str` DEFAULT: `''`
`include_in_schema`	Whether the endpoints appear in the OpenAPI schema: `False` (default): served, but absent from `/openapi.json` and the docs pages. `True`: documented like any other route, with the `/healthz` response model and the `503` response. The endpoints answer the same either way. This decides what the schema publishes, not what is reachable. TYPE: `bool` DEFAULT: `False`
`show_details`	Whether `/healthz` includes each check's verbose `details` field (versions, hostnames, pool stats, ...): `False` (default): details are stripped. Safe for public endpoints. `True`: details are always included. Use only if `/healthz` is private. `Depends(fn)` where `fn` returns `bool`: wires `fn` into FastAPI's DI graph, so `Depends` chains, `yield` cleanup, `Security`, `Request` injection, and async all work naturally. Return `True` to show details, `False` to strip them. Raising `HTTPException` blocks the endpoint, so return `False` instead when you want a soft strip. TYPE: `bool \| Depends` DEFAULT: `False`
`healthz_dependencies`	FastAPI dependencies applied to `/healthz`. A failing dependency blocks the entire endpoint (`401`/`403`). Use to hide `/healthz` from the public while leaving `/livez` and `/readyz` open to orchestrators and load balancers. Independent of `show_details`. TYPE: `list[Depends] \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`DependencyNotFoundError`	If `fastapi` is not installed.
`TypeError`	If `show_details` is neither a bool nor a `Depends(...)` value.

metrics_router

metrics_router(
    component: Metrics | None = None,
    *,
    prefix: str = "",
    path: str = "/metrics",
    include_in_schema: bool = False,
    dependencies: list[Depends] | None = None,
) -> APIRouter

Create a FastAPI router that serves Prometheus metrics.

Mounts GET {prefix}{path} (default GET /metrics) returning the Prometheus exposition format rendered from the component's collector registry. The active component must use the prometheus exporter. The endpoint stays out of the OpenAPI schema unless include_in_schema=True.

PARAMETER	DESCRIPTION
`component`	Metrics component whose Prometheus registry the endpoint renders. When omitted, the router resolves the default instance from the active `Grelmicro` app (`Grelmicro(uses=[Metrics(...)])`). TYPE: `Metrics \| None` DEFAULT: `None`
`prefix`	URL prefix for the metrics endpoint (e.g. '/api/v1'). TYPE: `str` DEFAULT: `''`
`path`	Path of the metrics endpoint under the prefix. TYPE: `str` DEFAULT: `'/metrics'`
`include_in_schema`	Whether the endpoint appears in the OpenAPI schema: `False` (default): served, but absent from `/openapi.json` and the docs pages. `True`: documented as returning the Prometheus exposition, which is text and not JSON. The endpoint answers the same either way. This decides what the schema publishes, not what is reachable. TYPE: `bool` DEFAULT: `False`
`dependencies`	FastAPI dependencies applied to the metrics endpoint. A failing dependency blocks the endpoint (`401`/`403`). Use to gate `/metrics` behind authentication. TYPE: `list[Depends] \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`DependencyNotFoundError`	If `fastapi` is not installed.

PARAMETER	DESCRIPTION
`ttl`	How long this route's response is served from the cache, in whole seconds or as a `timedelta`. A float is refused. Defaults to the `ttl` the registered `CachedResponses` carries. TYPE: `int \| timedelta \| None` DEFAULT: `None`
`shared`	On a route that requires a caller, serve one stored response to every caller the route admits, a credential included. Pass it only when the response is the same for each of them. Without it, a request carrying a credential is answered by the handler. TYPE: `bool` DEFAULT: `False`

PARAMETER	DESCRIPTION
`*limiters`	The limiters this route spends, in the order given. TYPE: `RateLimiter` DEFAULT: `()`
`cost`	Tokens one call of this route spends of each. TYPE: `int` DEFAULT: `1`
`max_wait`	Seconds a throttled call waits before it is refused. `0.0` refuses as soon as the budget is spent. TYPE: `float` DEFAULT: `0.0`
`key`	Builds the bucket key from the ASGI scope, replacing the resolved caller, the same as the middleware takes. Return `None` to leave a call unmetered. TYPE: `Callable[[Any], str \| None] \| None` DEFAULT: `None`
`trusted`	The proxies whose forwarded entries may be believed. Not needed when a middleware already resolved the caller. TYPE: `TrustedProxies \| None` DEFAULT: `None`
`legacy_headers`	Also send the superseded `X-RateLimit-` fields. TYPE:* `bool` DEFAULT: `False`

PARAMETER	DESCRIPTION
`app`	The FastAPI application to wire. TYPE: `FastAPI`
`errors`	The registered component that renders each rejection. TYPE: `ErrorResponses`

RAISES	DESCRIPTION
`TypeError`	If `scopes` is a single string.
`ValueError`	If a scope is not an OAuth scope token.

FastAPI

grelmicro.integrations.fastapi

CurrentPrincipal module-attribute

CurrentToken module-attribute

OptionalPrincipal module-attribute

Claims module-attribute

Conditional module-attribute

ConditionalRequired module-attribute

ConditionalRequest

if_match instance-attribute

if_none_match instance-attribute

check

fresh

CheckResultResponse

status instance-attribute

critical class-attribute instance-attribute

error instance-attribute

details instance-attribute

HealthzResponse

status instance-attribute

checks instance-attribute

install

install_error_responses

install_middleware

is_bound

error_response

CachedResponse

RateLimited

Authenticated

Anonymous

health_router

metrics_router

CurrentPrincipal `module-attribute`

CurrentToken `module-attribute`

OptionalPrincipal `module-attribute`

Claims `module-attribute`

Conditional `module-attribute`

ConditionalRequired `module-attribute`

if_match `instance-attribute`

if_none_match `instance-attribute`

status `instance-attribute`

critical `class-attribute` `instance-attribute`

error `instance-attribute`

details `instance-attribute`

status `instance-attribute`

checks `instance-attribute`