FastAPI¶

pip install "servicewright[fastapi]"

A full-featured HTTP entrypoint: the platform middleware stack, RFC 9457 error handling, Kubernetes probes, per-request DI scopes and a graceful drain — on a FastAPI app you still fully control.

from fastapi import APIRouter

from servicewright import AppSpec, Service
from servicewright.adapters.fastapi import FastApiEntrypoint, HttpConfig

router = APIRouter()


@router.get("/orders/{order_id}")
async def get_order(order_id: str) -> dict:
    return {"id": order_id}


http = FastApiEntrypoint(config=HttpConfig(port=8000), routers=(router,))
service = Service(spec, entrypoints=[http])

The /system namespace¶

Everything operational lives under one prefix, so it is trivial to keep off your public router:

These paths are excluded from request logging and from tracing by default.

Constructor¶

FastApiEntrypoint(
    config=HttpConfig(),
    routers=(),
    routes_registerer=None,
    middlewares=MiddlewareConfig(),
    exception_handlers=None,
    default_exception_handlers=True,
    error_renderer=None,
    metrics=False,
    configure_app=None,
    kind="http",
    essential=True,
)

HttpConfig¶

From settings¶

The environment-facing side of the same field set ships with the adapter: HttpServerSettings, one pydantic model per config dataclass — HealthSettings, MiddlewareSettings with its four middleware sections, UvicornSettings — same field names, same defaults, to_config() on each. Nest it in your settings object (servicewright[settings]'s BaseServiceSettings, or a BaseSettings of your own with env_nested_delimiter="__") and the mapping lives here rather than in each service:

from servicewright.adapters.fastapi import FastApiEntrypoint, HttpServerSettings
from servicewright.adapters.settings import BaseServiceSettings


class Settings(BaseServiceSettings):
    server: HttpServerSettings = HttpServerSettings()


settings = Settings()
http = FastApiEntrypoint(
    config=settings.server.to_config(version=settings.app_version),
    middlewares=settings.server.middlewares.to_config(),
    routers=(router,),
)

SERVER__PORT=8080
SERVER__GRACEFUL_TIMEOUT=15
SERVER__HEALTH__READINESS_PATH=/readyz
SERVER__UVICORN__TIMEOUT_KEEP_ALIVE=30
SERVER__UVICORN__FORWARDED_ALLOW_IPS=*
SERVER__MIDDLEWARES__GZIP__ENABLED=false
SERVER__MIDDLEWARES__CORS__ALLOW_ORIGINS='["https://app.example.com"]'

Every model is a plain BaseModel, never a BaseSettings, so the only way into server.port is SERVER__PORT: a bare PORT or HOST in the pod cannot reach it. HttpServerSettings().to_config() equals HttpConfig() and MiddlewareSettings().to_config() equals MiddlewareConfig(), and a test asserts that each model names exactly the fields of its dataclass minus the members below, so a field added to a dataclass cannot leave the mapping silently incomplete. MiddlewareConfig is the entrypoint's own argument rather than part of HttpConfig, hence the second call.

The members that are code, not environment, are arguments of to_config() rather than fields:

fastapi_kwargs and uvicorn_kwargs stay dict fields, JSON in the environment (SERVER__FASTAPI_KWARGS='{"debug": true}'); anything that is not JSON — a lifespan, say — goes onto the returned config afterwards, config.fastapi_kwargs["lifespan"] = lifespan.

port=0 is rejected unless allow_ephemeral_port=True, as on MetricsSettings: 0 only ever arrives here from configuration (SERVER__PORT, one keystroke from 8000), and a pod on an ephemeral port is one no probe or load balancer reaches. HttpConfig(port=0) itself keeps working and reports what it got through bound_port. CORS allow_credentials=True with a * origin fails at load, with server.middlewares.cors in the error, rather than at construction.

The uvicorn knobs¶

server.uvicorn types the operational uvicorn arguments. Every field is None until set and only a set field is forwarded, so uvicorn's own defaults apply to the rest and the model pins none of them; to_config() puts exactly what was set into HttpConfig.uvicorn_kwargs, typed — SERVER__UVICORN__TIMEOUT_KEEP_ALIVE=30 arrives as the integer 30, where the same key under a dict field would arrive as the string "30".

uvicorn_kwargs is the escape hatch for the rest of uvicorn.Config's signature and wins on a collision, as it does on HttpConfig itself. A test asserts every typed field is a uvicorn.Config parameter and lands there under its own name.

Not typed, on purpose, because the runner pre-binds the socket and drives one uvicorn.Server directly:

Not covered at all: cookie policy (secure, httponly, samesite) is an application concern and stays in your settings; MetricsInstrumentatorConfig is callables and library kwargs, wired in code as under Metrics; LitestarConfig has no settings model.

Per-request dependency scope¶

UnitScopeMiddleware is installed automatically as the outermost middleware and opens one unit scope per request. Reach it three ways:

from servicewright.adapters.fastapi import UnitScopeDep, current_unit_scope

# 1. The dependency (most common)
@router.get("/orders/{order_id}")
async def get_order(order_id: str, scope: UnitScopeDep) -> dict:
    use_case = await scope.get(GetOrder)
    return await use_case.execute(order_id)


# 2. Off the request
async def handler(request: Request):
    scope = request.state.unit_scope


# 3. From anywhere in the call tree
def deep_inside_something():
    scope = current_unit_scope()

Letting your DI integration own the scope¶

If your DI library's own FastAPI integration already opens a request scope — dishka's setup_dishka() with FromDishka / @inject handlers, for instance — switch the adapter's middleware off so the two never open two scopes per request:

from dishka.integrations.fastapi import setup_dishka

from servicewright.adapters.fastapi import FastApiEntrypoint, MiddlewareConfig


def configure_app(app: FastAPI, ctx: ServiceContext) -> None:
    setup_dishka(ctx.container.container, app)   # dishka's ContainerMiddleware, outermost


http = FastApiEntrypoint(
    routers=(router,),
    middlewares=MiddlewareConfig(unit_scope=False),
    configure_app=configure_app,
)

unit_scope=False drops UnitScopeMiddleware from the stack and nothing else: the context, logging, error and Sentry layers, the probes and the graceful drain do not depend on it. In this mode UnitScopeDep, request.state.unit_scope and current_unit_scope() raise LookupError — resolve through the integration that owns the scope. See dishka for the full picture, including what happens if both end up installed.

Middleware stack¶

from servicewright.adapters.fastapi import (
    CORSMiddlewareConfig,
    GZipMiddlewareConfig,
    LoggingMiddlewareConfig,
    MiddlewareConfig,
)

middlewares = MiddlewareConfig(
    unit_scope=True,
    context=True,
    sentry=True,
    processing_time=True,
    context_setters=None,
    logging=LoggingMiddlewareConfig(enabled=True, ignored_paths=["/system/health/livez"]),
    gzip=GZipMiddlewareConfig(enabled=True, minimum_size=1000),
    cors=CORSMiddlewareConfig(allow_origins=["https://app.example.com"]),
    custom=[(MyMiddleware, {"option": 1})],
)

Order, outermost first:

flowchart TD
    REQ["request"] --> A["UnitScope"]
    A --> B["context"]
    B --> C["unhandled-error"]
    C --> D["sentry"]
    D --> E["processing-time"]
    E --> F["logging"]
    F --> G["gzip"]
    G --> H["CORS"]
    H --> I["your custom middleware"]
    I --> J["handler"]

The request id¶

ContextMiddleware owns it. It takes the client's X-Request-ID when the value is log-safe, generates one otherwise, binds it into the context store and returns it in the response header.

Change the header name, or stop echoing:

from servicewright.adapters.fastapi import CorrelationIdMiddlewareConfig

MiddlewareConfig(correlation_id=CorrelationIdMiddlewareConfig(header_name="X-Correlation-ID"))
MiddlewareConfig(correlation_id=CorrelationIdMiddlewareConfig(enabled=False))   # bind, do not echo

Error handling¶

Five handlers are installed by default, all rendering through one pluggable renderer:

FastApiEntrypoint(
    error_renderer=MyEnvelopeRenderer(),           # own the wire format
    exception_handlers={LegacyError: my_handler},  # add your own types
    default_exception_handlers=False,              # or opt out entirely
)

Full details in Errors.

Metrics¶

FastApiEntrypoint(metrics=True)

Instruments the app with prometheus-fastapi-instrumentator and exposes /system/metrics on the API's own port, with /system/* excluded from instrumentation.

This is independent of the standalone exposition server that the Prometheus backend starts when settings.metrics.enabled is true. Both write into the same registry.

To tune the instrumentator, wire it yourself:

from servicewright.adapters.fastapi import MetricsInstrumentatorConfig, setup_metrics_instrumentator


def configure_app(app, ctx) -> None:
    setup_metrics_instrumentator(
        app,
        config=MetricsInstrumentatorConfig(init_kwargs={"should_group_status_codes": False}),
    )


FastApiEntrypoint(metrics=False, configure_app=configure_app)

MetricsInstrumentatorConfig is a pass-through, one field per call the wrapper makes: init_kwargs to Instrumentator(...), instrumentations to .add(...), instrument_kwargs to .instrument(...) and expose_kwargs to .expose(...).

instrumentations takes the library's per-request callables — its own metrics.request_size() and friends, or your own Callable[[Info], None | Awaitable[None]]. Anything passed there replaces the default set rather than extending it, and the two do not combine: metrics.default() already registers http_request_size_bytes / http_response_size_bytes (summaries labelled by handler only), so a request_size() added next to it — the same series with method and status labels — hits a duplicated time series and is silently dropped. To keep the built-ins alongside the finer-labelled sizes, compose the set from the pieces:

from prometheus_fastapi_instrumentator import metrics

MetricsInstrumentatorConfig(
    instrumentations=[
        metrics.requests(),
        metrics.latency(),
        metrics.request_size(),
        metrics.response_size(),
    ],
)

Tracing¶

pip install "servicewright[fastapi-tracing]"

With ObsConfig(tracing="otel") and a configured settings.tracing, the entrypoint asks the tracing sink to instrument the app at bind time — before the middleware stack, so spans wrap the whole chain. The /system/* paths and logging.ignored_paths are excluded automatically, on top of anything in settings.tracing.excluded_urls.

With tracing off, this is a no-op. Nothing is imported.

Registering routes¶

Three ways, in the order they are applied:

# 1. Routers
FastApiEntrypoint(routers=(orders_router, users_router))


# 2. A registerer, when routes need the ServiceContext
async def register_routes(app: FastAPI, ctx: ServiceContext) -> None:
    catalog = await ctx.app_scope.get(Catalog)
    app.include_router(build_router(catalog))


FastApiEntrypoint(routes_registerer=register_routes)


# 3. The final hook, after everything else is wired
def configure_app(app: FastAPI, ctx: ServiceContext) -> None:
    app.mount("/static", StaticFiles(directory="static"))


FastApiEntrypoint(configure_app=configure_app)

Lifecycle notes¶

The app has no lifespan that manages your container. The Host already ran bootstrap, the application scope, warmup and pre_start before bind() is called.

• bind() builds the app and opens the listening socket.
• serve() runs uvicorn with its signal capture neutralized, and returns — still accepting — when the Host's stop event is set.
• drain(grace) closes the listener and lets in-flight requests finish.
• stop() is the hard stop.

http = FastApiEntrypoint(config=HttpConfig(port=0))
# after bind:
http.bound_port   # e.g. 54312
http.app          # the built FastAPI instance, or None before bind

Testing¶

Build the app without running a server:

from servicewright.testing import FakeContainer, FakeSettings

ctx = ServiceContext(bootstrap=..., app_scope=..., health=HealthRegistry())
app = await FastApiEntrypoint(routers=(router,)).build_app(ctx)

transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
    response = await client.get("/orders/42")

See Testing for the full setup.

Handy header types¶

Common headers as ready-made annotated dependencies:

from servicewright.adapters.fastapi import (
    AuthorizationHeader,   # Authorization
    IdempotencyKey,        # Idempotency-Key, validated and length-capped
    XFingerprintHeader,    # X-Fingerprint
    XUserId,               # X-User-Id, parsed as UUID
)


@router.post("/orders")
async def create_order(user_id: XUserId, idempotency_key: IdempotencyKey) -> dict: ...