FastAPI

• Start here: Request handlers and the ambient scope
• Common recipes: app.add_middleware(GrelmicroMiddleware, micro=micro) binds the active app inside request handlers, so patterns resolve their backends ambiently without explicit backend= wiring. health_router() adds the /livez, /readyz, and /healthz endpoints. app.add_middleware(IdempotencyMiddleware, idempotency=Idempotency("http")) replays a stored response when a request repeats its Idempotency-Key.

grelmicro.integrations.fastapi

FastAPI integration: middleware, install helper, and health router.

GrelmicroMiddleware

GrelmicroMiddleware(app: ASGIApp, *, micro: Grelmicro)

Bind the active Grelmicro app for the duration of each request.

A request handler runs in its own task, outside the async with micro: block, so Grelmicro.current() and the ambient backend= resolution it powers do not see the app there. This middleware sets the active app for the request task, so Lock("cart"), RateLimiter.sliding_window(...), and @cached resolve ambiently inside the handler exactly as they do in a task.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.integrations.fastapi import GrelmicroMiddleware

micro = Grelmicro(uses=[...])

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with micro:
        yield

app = FastAPI(lifespan=lifespan)
app.add_middleware(GrelmicroMiddleware, micro=micro)

Open the app in the framework lifespan so its components are registered before any request arrives. The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It binds on http and websocket scopes and passes the lifespan scope through untouched.

Initialize the middleware with the app to bind.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
micro	The Grelmicro app to bind for each request. Open it in the framework lifespan so its components are ready. TYPE: Grelmicro

app instance-attribute

app = app

micro instance-attribute

micro = micro

IdempotencyMiddleware

IdempotencyMiddleware(
    app: ASGIApp,
    *,
    idempotency: Idempotency[Any],
    header: str = "Idempotency-Key",
    methods: Collection[str] = ("POST",),
    key_maker: Callable[[Scope, str], str] | None = None,
    skip: Callable[[StoredResponse], bool] | None = None,
    require_key: bool = False,
    fingerprint_body: bool = False,
    max_body_size: int = 1024 * 1024,
    wait_timeout: float = 10.0,
)

Replay a stored HTTP response when a request repeats its idempotency key.

A request whose method is listed in methods and which carries the header runs once. A retry with the same key replays the stored status, headers, and body without reaching the handler, and carries idempotent-replayed: true. A request without the header passes straight through, so adding the middleware changes nothing until a client opts in.

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.idempotency import Idempotency
from grelmicro.integrations.fastapi import IdempotencyMiddleware

micro = Grelmicro(uses=[...])
app = FastAPI()
micro.install(app)

app.add_middleware(
    IdempotencyMiddleware, idempotency=Idempotency("http", ttl=3600)
)

Add it before or after micro.install(app). It resolves its Cache through the grelmicro request scope, which install keeps outside every other middleware.

A duplicate that arrives while the first execution is in flight waits for it and replays its response. The wait folds across replicas when a Coordination lock backend is configured, and in-process otherwise. It is bounded by wait_timeout.

Every response the app returns is stored, errors included. A handler that raises an unhandled exception stores nothing, so the framework's 500 never replays.

Four kinds of response are not stored, and each one lets a retry re-run the handler: one carrying Set-Cookie, one carrying Content-Encoding, one declaring trailers, and one whose body is over max_body_size. All four are logged. Pass skip to add a rule of your own.

Background tasks run after the response is sent, so a replay can be served while the original request's background work is still in flight.

The middleware is pure ASGI and works with any ASGI framework (Starlette, Litestar, ...). It acts on http scopes and passes every other scope through untouched.

Initialize the middleware with the idempotency store and policy.

PARAMETER	DESCRIPTION
app	The next ASGI application in the middleware chain. TYPE: ASGIApp
idempotency	The Idempotency that stores responses. Its ttl sets how long a key replays. TYPE: Idempotency[Any]
header	Request header carrying the idempotency key. TYPE: str DEFAULT: 'Idempotency-Key'
methods	Methods that take an idempotency key. Every other method passes through. TYPE: Collection[str] DEFAULT: ('POST',)
key_maker	Build the stored key from the ASGI scope and the client key. Defaults to the method, the path, the query string, and the client key, so two routes never replay each other. Set this in any multi-tenant app, folding in the caller identity. Without it a client that learns another client's key replays their response. TYPE: Callable[[Scope, str], str] | None DEFAULT: None
skip	Predicate receiving the response. Return True to not store it. Mirrors skip on @cached. Use it for a response that is technically replayable but should not be, such as one whose body embeds a timestamp the caller must not see twice. Responses that are never safe to replay are dropped before this runs. TYPE: Callable[[StoredResponse], bool] | None DEFAULT: None
require_key	Answer 400 when a method in methods arrives without the header, instead of passing it through. TYPE: bool DEFAULT: False
fingerprint_body	Hash the request body and store the hash with the response. A key reused with a different body then gets 422 instead of a wrong replay. Buffers the request body before the handler runs, and answers 413 when it is over max_body_size. TYPE: bool DEFAULT: False
max_body_size	Largest body held in memory, in bytes. A larger response is sent to the client and not stored. With fingerprint_body, a larger request body is answered with 413. TYPE: int DEFAULT: 1024 * 1024
wait_timeout	Seconds a duplicate waits for an execution already in flight. Past it the duplicate is answered with 409 and a Retry-After header. TYPE: float DEFAULT: 10.0

app instance-attribute

app = app

StoredResponse

Bases: TypedDict

The response IdempotencyMiddleware is about to store.

Handed to skip so a handler's own rule decides whether a response replays. headers maps lowercased names to their value, keeping the last of a repeated name.

status instance-attribute

status: int

headers instance-attribute

headers: dict[str, str]

body instance-attribute

body: bytes

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]

document_idempotency

document_idempotency(app: FastAPI) -> None

Describe the installed IdempotencyMiddleware in the OpenAPI schema.

A middleware runs outside the routing layer, so nothing it does reaches the generated schema and a client built from that schema never learns the header exists. This reads the installed middleware and annotates every operation it covers with the header parameter and the responses the middleware itself can return.

from grelmicro.integrations.fastapi import (
    IdempotencyMiddleware,
    document_idempotency,
)

app.add_middleware(IdempotencyMiddleware, idempotency=Idempotency("http"))
micro.install(app)
document_idempotency(app)

Call it any time after add_middleware. The schema is annotated the next time it is built, so routes added afterwards are covered too.

An operation that already declares the header keeps its own declaration. A 422 that FastAPI generated for request validation keeps its schema, and the idempotency case is added to its description.

A mounted sub-application builds its own schema, which this does not reach. Call it on the sub-application as well.

PARAMETER	DESCRIPTION
app	The app carrying an IdempotencyMiddleware to document. TYPE: FastAPI

RAISES	DESCRIPTION
DependencyNotFoundError	If fastapi is not installed.
TypeError	If app is not a FastAPI app, or carries no IdempotencyMiddleware.

health_router

health_router(
    component: HealthChecks | None = None,
    *,
    prefix: str = "",
    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. Never runs checkers. Always returns 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.

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: ''
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.