Skip to content

Errors and observability

Problem details, request IDs, request deadlines, health checks and logs as JSON. In fastapi_alcazaba.observability.

In the guide: Problem details, Request IDs, Request timeouts, Health checks, Structured logs. In an app: the example.

fastapi_alcazaba.observability.FastAPIProblemDetails 

FastAPIProblemDetails(*, expose_detail: bool = True, max_errors: int = 100)

Answers errors with RFC 9457 problem details. install(app) handles:

  • HTTPException (FastAPI's and Starlette's, and subclasses such as BreakerOpen and Problem), keeping its headers (Retry-After, WWW-Authenticate...). A 4xx's detail is sent, since it's written for clients (expose_detail=False keeps it); a 5xx's never is, except a Problem's. Codes under 400 (redirects raised as exceptions) keep FastAPI's answer.
  • RequestValidationError: a 422 whose errors list each error's loc, msg and type, never its input (which echoes what was sent, passwords included) nor its ctx.

  • any other exception: a 500 that says nothing about it. Starlette raises the exception again after the response, so that the server logs it with its traceback (uvicorn's "Exception in ASGI application") and TestClient raises it in tests (raise_server_exceptions=False to see the response). With debug=True, Starlette answers with its traceback page instead.

    FastAPIProblemDetails().install(app)

Parameters:

Name Type Description Default
expose_detail bool

send the detail of 4xx HTTPExceptions.

 True
max_errors int

validation errors listed at most, so that a huge invalid body doesn't get a huge answer.

 100

install 

install(app: Starlette) -> None

Register the handlers in app, before it starts. A handler you register for a status code or an exception subclass still wins for those. Alcazaba's middlewares answer with problems too (the 504 of a timeout, the 413 of a body too large...).

response 

response(request: Request, exc: HTTPException) -> Response

The answer to exc, for your own exception handlers:

@app.exception_handler(OutOfStock)
async def out_of_stock(request: Request, exc: OutOfStock) -> Response:
    return problems.response(request, Problem(409, title="Out of stock"))

The same rules apply as for a raised exception.

fastapi_alcazaba.observability.Problem 

Problem(status: int, detail: str | None = None, *, type: str = 'about:blank', title: str | None = None, headers: Mapping[str, str] | None = None, extensions: Mapping[str, Any] | None = None)

Bases: HTTPException

An HTTPException with the members of an RFC 9457 problem: its type (a URI that identifies the kind of problem, ideally to its documentation), title, detail and extension members. Its detail and extensions always reach the client, also with a 5xx: you write them for it.

raise Problem(403, "Your balance is 30, but that costs 50",
              type="https://example.com/problems/out-of-credit", title="Out of credit",
              extensions={"balance": 30, "cost": 50})

Parameters:

Name Type Description Default
status int

a 4xx or 5xx code.

 required
detail str | None

what happened this time, for the client to read.

 None
type str

the problem type's URI; "about:blank" means the status says it all.

 'about:blank'
title str | None

a short summary of the problem type; the status phrase by default.

 None
headers Mapping[str, str] | None

response headers, such as Retry-After.

 None
extensions Mapping[str, Any] | None

more members, with names of three or more letters, digits or "_" (starting with a letter) and values that FastAPI can encode as JSON; they can't replace the standard members.

 None

fastapi_alcazaba.observability.RequestIDMiddleware 

RequestIDMiddleware(app: ASGIApp, *, header: str = 'X-Request-ID', trust_incoming: bool = True)

Gives every HTTP request and WebSocket an ID, available from request_id() while it's handled and sent back in the response's header.

app.add_middleware(RequestIDMiddleware, trust_incoming=False)

Add it last, the outermost, so that every log line and every answer of the other middlewares carry the ID. When the app raises, the ID stays set for the rest of the request's task: the error handler and the server log the exception outside every middleware.

Parameters:

Name Type Description Default
header str

the request and response header with the ID.

 'X-Request-ID'
trust_incoming bool

reuse the ID in the request's header (when safe; a single value). Set it to False when clients reach the app without a proxy that sets the header itself and you don't want them to choose their IDs.

 True

fastapi_alcazaba.observability.request_id

A request ID for every request, to find all the log lines of one request and to let a client quote it when it reports an error.

app.add_middleware(RequestIDMiddleware)  # last: the outermost

handler.addFilter(RequestIDLogFilter())  # then "%(request_id)s" in the format

RequestIDMiddleware takes the ID from the request's X-Request-ID (set by a proxy or the client) when it's safe to log, else makes a random one; request_id() returns it anywhere while the request is handled, and the response carries it in the same header. An incoming ID is reused only if it matches [A-Za-z0-9._:-]{1,128}: anything else (line breaks that would forge log lines, an endless value) gets a new one.

RequestIDMiddleware 

RequestIDMiddleware(app: ASGIApp, *, header: str = 'X-Request-ID', trust_incoming: bool = True)

Gives every HTTP request and WebSocket an ID, available from request_id() while it's handled and sent back in the response's header.

app.add_middleware(RequestIDMiddleware, trust_incoming=False)

Add it last, the outermost, so that every log line and every answer of the other middlewares carry the ID. When the app raises, the ID stays set for the rest of the request's task: the error handler and the server log the exception outside every middleware.

Parameters:

Name Type Description Default
header str

the request and response header with the ID.

 'X-Request-ID'
trust_incoming bool

reuse the ID in the request's header (when safe; a single value). Set it to False when clients reach the app without a proxy that sets the header itself and you don't want them to choose their IDs.

 True

RequestIDLogFilter 

RequestIDLogFilter(*, default: str = '-')

Bases: Filter

Sets record.request_id to the ID of the request being handled (default outside one), for formats with %(request_id)s. Add it to the handlers: a logger's filters don't see the records of its child loggers.

handler.addFilter(RequestIDLogFilter())

A request_id passed in extra= is kept.

request_id 

request_id() -> str | None

The ID of the request being handled; None outside a request, or without the middleware.

fastapi_alcazaba.observability.RequestIDLogFilter 

RequestIDLogFilter(*, default: str = '-')

Bases: Filter

Sets record.request_id to the ID of the request being handled (default outside one), for formats with %(request_id)s. Add it to the handlers: a logger's filters don't see the records of its child loggers.

handler.addFilter(RequestIDLogFilter())

A request_id passed in extra= is kept.

fastapi_alcazaba.observability.RequestTimeoutMiddleware 

RequestTimeoutMiddleware(app: ASGIApp, *, timeout: float | None = 30.0, paths: Mapping[str, float | None] | None = None)

Gives every HTTP request timeout seconds, from its arrival until its response has been sent; WebSockets and lifespan events go through untouched.

app.add_middleware(RequestTimeoutMiddleware, timeout=30, paths={"/exports": None})
  • Out of time before the response starts: the app is cancelled and the client gets a 504 with a small JSON body.
  • Out of time while a response streams: too late for a 504. The app is cancelled and the exception (a TimeoutError) raised again, so that the server closes the connection: the client sees an incomplete body, never a short one that looks complete. Give streaming endpoints (Server-Sent Events, big exports) their own limit in paths, or None.
  • Once the response has been sent, background tasks have no deadline.
  • Sync endpoints run in a thread that can't be interrupted: the client gets its 504 on time, but the thread runs to its end, after the exit of its yield dependencies has run. Endpoints that may hit the deadline are better async.

Both cases are logged as warnings.

Parameters:

Name Type Description Default
timeout float | None

the deadline, in seconds; None for none.

 30.0
paths Mapping[str, float | None] | None

other deadlines for the paths starting with these prefixes (the longest matching prefix wins; None for none), matched like the app's routes: without the root_path a proxy or a mount adds in front.

 None

timeout_for 

timeout_for(path: str) -> float | None

The deadline of requests to path, in seconds; None when there's none.

fastapi_alcazaba.observability.health_router 

health_router(checks: Mapping[str, Check | HealthCheck] | None = None, *, cache: Cache | None = None, include_cache: bool = True, prefix: str = '/health', timeout: float = 2.0, reuse_for: float = 1.0, include_in_schema: bool = False, tags: list[str | Enum] | None = None) -> APIRouter

GET {prefix}/live and GET {prefix}/ready, to include in the app.

app.include_router(health_router({"database": check_database}))

Parameters:

Name Type Description Default
checks Mapping[str, Check | HealthCheck] | None

readiness checks by name: callables (critical) or HealthChecks.

 None
cache Cache | None

the Cache whose backend /ready checks; else the one get_cache returns, if any is configured.

 None
include_cache bool

add the non-critical "cache" check.

 True
prefix str

path prefix of both endpoints ("" for none).

 '/health'
timeout float

seconds each check may take, unless its HealthCheck says otherwise.

 2.0
reuse_for float

seconds during which a result answers the next probes; 0 runs the checks for every probe (concurrent probes still share one run).

 1.0
include_in_schema bool

show the endpoints in the OpenAPI schema.

 False
tags list[str | Enum] | None

OpenAPI tags of the endpoints.

 None

fastapi_alcazaba.observability.HealthCheck dataclass 

HealthCheck(check: Check, critical: bool = True, timeout: float | None = None)

A readiness check with its options.

Parameters:

Name Type Description Default
check Check

sync or async callable; it passes when it returns anything but False without raising within timeout. Sync ones run in a thread of their own, off the event loop.

 required
critical bool

when it fails, the app isn't ready (503); a non-critical one only makes it "degraded" (200), for dependencies the app can work without.

 True
timeout float | None

seconds it may take; the router's timeout by default.

 None

fastapi_alcazaba.observability.JSONFormatter 

JSONFormatter(*, static: Mapping[str, Any] | None = None, redact: Collection[str] = ())

Bases: Formatter

Formats each record as one line of JSON, ASCII only: nothing in it splits a line, a message with line breaks included.

Parameters:

Name Type Description Default
static Mapping[str, Any] | None

fields every line has, such as {"service": "shop", "environment": "prod"}.

 None
redact Collection[str]

words that mark secrets in the keys of extra, besides the audit trail's SENSITIVE_WORDS.

 ()

fastapi_alcazaba.observability.logging_config 

logging_config(level: str = 'INFO', *, static: Mapping[str, Any] | None = None, redact: Collection[str] = (), access_log: bool = True) -> dict[str, Any]

A logging.config.dictConfig that sends every record, Uvicorn's too, to standard output as JSON: for uvicorn.run(..., log_config=logging_config()), or dictConfig() at start.

Parameters:

Name Type Description Default
level str

of the app's loggers and the root.

 'INFO'
static Mapping[str, Any] | None

fields every line has ({"service": "shop"}).

 None
redact Collection[str]

more words that mark secrets in extra.

 ()
access_log bool

keep Uvicorn's line per request.

 True