API reference: adapters¶

class DaemonEntrypoint(ScopedEntrypoint):
    """Runs ``func(scope, stop)`` in one long-lived unit scope.

    The function is expected to loop until ``stop`` is set.
    """

    def __init__(
        self,
        func: Callable[[UnitScopeProtocol, asyncio.Event], Awaitable[None]],
        *,
        kind: str = "daemon",
        essential: bool = True,
    ) -> None:
        super().__init__()
        self._func = func
        self.kind = kind
        self.essential = essential

    async def serve(self, *, stop: asyncio.Event) -> None:
        """Open one unit scope and hand control to the user loop."""
        async with self.unit_scope() as scope:
            await self._func(scope, stop)

class OneShotEntrypoint(ScopedEntrypoint):
    """Runs ``func`` exactly once inside a fresh unit scope, then returns.

    Being ``essential`` by default, its return stops the whole service.
    """

    def __init__(
        self,
        func: Callable[[UnitScopeProtocol], Awaitable[None]],
        *,
        kind: str = "oneshot",
        essential: bool = True,
    ) -> None:
        super().__init__()
        self._func = func
        self.kind = kind
        self.essential = essential

    async def serve(self, *, stop: asyncio.Event) -> None:
        """Open a unit scope, run the function once, then return."""
        async with self.unit_scope() as scope:
            await self._func(scope)

@dataclass(slots=True)
class CORSMiddlewareConfig:
    """Configuration for Starlette's ``CORSMiddleware``."""

    enabled: bool = True
    allow_origins: list[str] = field(default_factory=list)
    allow_credentials: bool = False
    allow_methods: list[str] = field(default_factory=lambda: list(DEFAULT_CORS_METHODS))
    allow_headers: list[str] = field(default_factory=lambda: ["*"])
    expose_headers: list[str] = field(default_factory=list)
    max_age: int = 600

    def __post_init__(self) -> None:
        """Reject the insecure ``allow_credentials`` + wildcard-origin combo."""
        if self.allow_credentials and "*" in self.allow_origins:
            raise ValueError(
                "allow_credentials=True cannot be used with allow_origins=['*']. Specify explicit origins for security."
            )

class CORSMiddlewareSettings(BaseModel):
    """``CORSMiddlewareConfig`` as settings; the lists take JSON lists (``ALLOW_ORIGINS='["https://app.example.com"]'``)."""

    enabled: bool = Field(default=True, description="Install Starlette's CORSMiddleware")
    allow_origins: list[str] = Field(default_factory=list, description="Origins allowed; empty allows none")
    allow_credentials: bool = Field(default=False, description="Allow cookies and Authorization across origins")
    allow_methods: list[str] = Field(default_factory=lambda: list(DEFAULT_CORS_METHODS), description="Methods allowed")
    allow_headers: list[str] = Field(default_factory=lambda: ["*"], description="Request headers allowed")
    expose_headers: list[str] = Field(default_factory=list, description="Response headers exposed to the browser")
    max_age: int = Field(default=600, ge=0, description="Seconds a preflight response may be cached")

    @model_validator(mode="after")
    def _validate(self) -> CORSMiddlewareSettings:
        # The dataclass owns the credentials-with-wildcard rule; raising it here puts the field path on it.
        self.to_config()
        return self

    def to_config(self) -> CORSMiddlewareConfig:
        return CORSMiddlewareConfig(
            enabled=self.enabled,
            allow_origins=list(self.allow_origins),
            allow_credentials=self.allow_credentials,
            allow_methods=list(self.allow_methods),
            allow_headers=list(self.allow_headers),
            expose_headers=list(self.expose_headers),
            max_age=self.max_age,
        )

@dataclass(slots=True)
class CorrelationIdMiddlewareConfig:
    """How the request id is returned to the client.

    The id itself is owned by :class:`ContextMiddleware` — the same value that
    lands in the context store, the logs and outbound propagation. Echoing it
    is what lets a caller quote an id that actually appears in your logs.
    """

    enabled: bool = True
    header_name: str = DEFAULT_CORRELATION_ID_HEADER

class CorrelationIdMiddlewareSettings(BaseModel):
    """``CorrelationIdMiddlewareConfig`` as settings: how the request id is echoed."""

    enabled: bool = Field(default=True, description="Echo the request id in a response header")
    header_name: str = Field(default=DEFAULT_CORRELATION_ID_HEADER, description="The header it is echoed in")

    def to_config(self) -> CorrelationIdMiddlewareConfig:
        return CorrelationIdMiddlewareConfig(enabled=self.enabled, header_name=self.header_name)

class FastApiEntrypoint(ServerEntrypoint):
    """A FastAPI server entrypoint driven by the :class:`Host`.

    Args:
        config: Self-contained server configuration (NOT read from settings).
        routers: APIRouter instances to ``include_router`` onto the app.
        routes_registerer: Optional callback to register routes imperatively,
            resolved at ``bind`` time with the :class:`ServiceContext`.
        middlewares: Middleware stack configuration.
        exception_handlers: Extra ``{exc_type: handler}`` mappings appended after
            the default handlers.
        default_exception_handlers: Install the default handlers
            (validation/HTTP/ServiceError/deadline/unhandled). Default ``True``.
        error_renderer: Wire-format renderer used by the default handlers;
            ``None`` = RFC 9457 ``ProblemDetailsRenderer``. Pass your own
            implementation to own the error format (custom envelope,
            localization) across every default handler.
        metrics: Expose in-app Prometheus metrics at ``/system/metrics`` via
            ``prometheus-fastapi-instrumentator`` (requires servicewright[fastapi]).
        configure_app: Final hook called with ``(app, ctx)`` after wiring.
        kind: Telemetry label (default ``"http"``).
        essential: Whether the entrypoint's exit/failure stops the process.
    """

    def __init__(
        self,
        *,
        config: HttpConfig | None = None,
        routers: tuple[Any, ...] = (),
        routes_registerer: RoutesRegisterer | None = None,
        middlewares: MiddlewareConfig | None = None,
        exception_handlers: dict[type[Exception], ExceptionHandler] | None = None,
        default_exception_handlers: bool = True,
        error_renderer: HttpErrorRendererProtocol | None = None,
        metrics: bool = False,
        configure_app: ConfigureApp | None = None,
        kind: str = "http",
        essential: bool = True,
    ) -> None:
        self._config = config if config is not None else HttpConfig()
        self._routers = tuple(routers)
        self._routes_registerer = routes_registerer
        self._middlewares = middlewares if middlewares is not None else MiddlewareConfig()
        self._exception_handlers: dict[type[Exception], ExceptionHandler] = dict(exception_handlers or {})
        self._default_exception_handlers = default_exception_handlers
        self._error_renderer = error_renderer
        self._metrics = metrics
        self._configure_app = configure_app
        self.kind = kind
        self.essential = essential

        self._app: FastAPI | None = None
        self._runner = UvicornRunner(
            host=self._config.host,
            port=self._config.port,
            graceful_timeout=self._config.graceful_timeout,
            uvicorn_kwargs=self._config.uvicorn_kwargs,
            label="FastAPI",
        )

    @property
    def config(self) -> HttpConfig:
        """The server configuration."""
        return self._config

    @property
    def app(self) -> FastAPI | None:
        """The built FastAPI application (``None`` before :meth:`bind`)."""
        return self._app

    @property
    def bound_port(self) -> int | None:
        """The actually bound port (useful when ``config.port == 0``)."""
        return self._runner.bound_port

    async def bind(self, ctx: ServiceContext[Any, Any]) -> None:
        """Build the FastAPI app and open the listening socket.

        The socket is opened here, not in :meth:`serve`, so a port clash aborts
        startup with an ``OSError`` while readiness is still false — instead of
        a process that reports ready and serves nothing. It also makes
        ``port=0`` usable: :attr:`bound_port` reports what the OS picked.
        """
        self._app = await self.build_app(ctx)
        bound_port = self._runner.bind()
        logger.info(
            "FastAPI entrypoint bound",
            extra={"service": ctx.service_name, "address": self._config.address, "port": bound_port},
        )

    async def build_app(self, ctx: ServiceContext[Any, Any]) -> FastAPI:
        """Construct a fully-configured FastAPI app from the :class:`ServiceContext`.

        Exposed for testability: callers can build the app without driving the
        serve loop. The app deliberately has NO container-managing lifespan —
        the Host owns the application scope.
        """
        app = self._create_app(ctx)

        if self._config.health.enabled:
            setup_health_routes(
                app,
                ctx.health,
                liveness_path=self._config.health.liveness_path,
                readiness_path=self._config.health.readiness_path,
            )

        setup_exception_handlers(
            app,
            setup_defaults=self._default_exception_handlers,
            error_renderer=self._error_renderer,
            custom_handlers=self._exception_handlers,
        )

        # OTel must instrument the app BEFORE middleware so it wraps the stack.
        instrument_fastapi_app(app, ctx, middlewares=self._middlewares, health=self._config.health)

        setup_middleware_stack(app, self._middlewares, ctx.container, error_renderer=self._error_renderer)

        for router in self._routers:
            app.include_router(router)

        await self._register_routes(app, ctx)

        if self._metrics:
            setup_metrics_instrumentator(app)

        if self._configure_app is not None:
            self._configure_app(app, ctx)

        return app

    async def serve(self, *, stop: asyncio.Event) -> None:
        """Serve on the bound socket until the host's ``stop`` event is set.

        Returns while the server is still accepting connections: the Host flips
        readiness to false first and only then calls :meth:`drain`, which is
        what closes the listener. Shutting uvicorn down here instead would make
        the readiness endpoint die before the load balancer stops routing, and
        would render the Host's drain grace meaningless.

        A server that dies on its own (a fatal uvicorn error) ends the wait and
        the failure is re-raised, so an essential entrypoint cannot leave the
        process alive and idle.
        """
        if self._app is None:
            raise RuntimeError("serve() called before bind()")
        await self._runner.serve(self._app, stop=stop)

    async def drain(self, grace: float) -> None:
        """Close the listener and let in-flight requests finish within ``grace``."""
        await self._runner.drain(grace)

    async def stop(self) -> None:
        """Hard stop the server immediately (idempotent with :meth:`drain`)."""
        await self._runner.stop()

    def _create_app(self, ctx: ServiceContext[Any, Any]) -> FastAPI:
        params: dict[str, Any] = {
            "title": self._config.title or ctx.service_name,
            "version": self._config.version,
            "openapi_url": self._config.openapi_url,
            "docs_url": self._config.docs_url,
            "redoc_url": self._config.redoc_url,
            "redirect_slashes": self._config.redirect_slashes,
        }
        params.update(self._config.fastapi_kwargs)
        return FastAPI(**params)

    async def _register_routes(self, app: FastAPI, ctx: ServiceContext[Any, Any]) -> None:
        if self._routes_registerer is None:
            return
        result = self._routes_registerer(app, ctx)
        if asyncio.iscoroutine(result):
            await result

class FastApiPlugin:
    """Declarative wiring: register a :class:`FastApiEntrypoint` on the host.

    Pass the same arguments as :class:`FastApiEntrypoint`; ``on_register`` builds
    it and adds it to the host.
    """

    def __init__(
        self,
        *,
        config: HttpConfig | None = None,
        routers: tuple[Any, ...] = (),
        routes_registerer: RoutesRegisterer | None = None,
        middlewares: MiddlewareConfig | None = None,
        exception_handlers: dict[type[Exception], ExceptionHandler] | None = None,
        default_exception_handlers: bool = True,
        error_renderer: HttpErrorRendererProtocol | None = None,
        metrics: bool = False,
        configure_app: ConfigureApp | None = None,
        kind: str = "http",
        essential: bool = True,
    ) -> None:
        self._entrypoint = FastApiEntrypoint(
            config=config,
            routers=routers,
            routes_registerer=routes_registerer,
            middlewares=middlewares,
            exception_handlers=exception_handlers,
            default_exception_handlers=default_exception_handlers,
            error_renderer=error_renderer,
            metrics=metrics,
            configure_app=configure_app,
            kind=kind,
            essential=essential,
        )

    @property
    def entrypoint(self) -> FastApiEntrypoint:
        """The entrypoint that will be registered on the host."""
        return self._entrypoint

    def on_register(self, spec: Any, host: Any) -> None:
        """Append the FastAPI entrypoint to the host."""
        host.add_entrypoint(self._entrypoint)

@dataclass(slots=True)
class GZipMiddlewareConfig:
    """Configuration for Starlette's ``GZipMiddleware``."""

    enabled: bool = True
    minimum_size: int = 1000

class GZipMiddlewareSettings(BaseModel):
    """``GZipMiddlewareConfig`` as settings."""

    enabled: bool = Field(default=True, description="Compress responses")
    minimum_size: int = Field(default=1000, ge=0, description="Smallest response body compressed, bytes")

    def to_config(self) -> GZipMiddlewareConfig:
        return GZipMiddlewareConfig(enabled=self.enabled, minimum_size=self.minimum_size)

@dataclass(slots=True)
class HealthConfig:
    """Configuration for the ``/system`` health probe routes."""

    enabled: bool = True
    liveness_path: str = DEFAULT_LIVENESS_PATH
    readiness_path: str = DEFAULT_READINESS_PATH

class HealthSettings(BaseModel):
    """``HealthConfig`` as settings: the ``/system`` probe routes."""

    enabled: bool = Field(default=True, description="Serve the liveness and readiness routes")
    liveness_path: str = Field(default=DEFAULT_LIVENESS_PATH, description="Liveness probe path")
    readiness_path: str = Field(default=DEFAULT_READINESS_PATH, description="Readiness probe path")

    def to_config(self) -> HealthConfig:
        return HealthConfig(enabled=self.enabled, liveness_path=self.liveness_path, readiness_path=self.readiness_path)

@dataclass(slots=True)
class HttpConfig:
    """Self-contained configuration for an HTTP server entrypoint.

    Taken at construction, never read from global settings, so the AppSpec
    stays transport-neutral.
    """

    host: str = DEFAULT_HTTP_HOST
    port: int = DEFAULT_HTTP_PORT
    graceful_timeout: float = DEFAULT_GRACEFUL_TIMEOUT_SECONDS

    # FastAPI app construction.
    title: str | None = None
    version: str = "0.0.0"
    openapi_url: str = DEFAULT_OPENAPI_URL
    docs_url: str = DEFAULT_DOCS_URL
    redoc_url: str = DEFAULT_REDOC_URL
    redirect_slashes: bool = False
    fastapi_kwargs: dict[str, Any] = field(default_factory=dict)

    # uvicorn server construction.
    uvicorn_kwargs: dict[str, Any] = field(default_factory=dict)

    # Components.
    health: HealthConfig = field(default_factory=HealthConfig)

    @property
    def address(self) -> str:
        """Return the ``host:port`` bind address."""
        return f"{self.host}:{self.port}"

class HttpServerSettings(BaseModel):
    """``HttpConfig`` as settings, plus the ``middlewares`` section the entrypoint takes beside it.

    ``port=0`` is rejected unless ``allow_ephemeral_port`` says so: ``0`` only ever
    arrives here from configuration, and a pod on an ephemeral port is one no
    probe or load balancer reaches. ``version`` is the argument of
    :meth:`to_config`, not a field: it is the application's version, which the
    settings object carries once already.
    """

    host: str = Field(default=DEFAULT_HTTP_HOST, description="Bind host")
    port: int = Field(default=DEFAULT_HTTP_PORT, ge=0, le=65535, description="Bind port")
    allow_ephemeral_port: bool = Field(
        default=False,
        description="Accept port=0 (ephemeral); off by default, a probe targeting a fixed port cannot reach one",
    )
    graceful_timeout: float = Field(
        default=DEFAULT_GRACEFUL_TIMEOUT_SECONDS,
        description="Seconds uvicorn lets in-flight requests finish on shutdown; 0 waits indefinitely",
    )
    title: str | None = Field(default=None, description="OpenAPI title; None falls back to AppSpec.service_name")
    openapi_url: str = Field(default=DEFAULT_OPENAPI_URL, description="Schema path")
    docs_url: str = Field(default=DEFAULT_DOCS_URL, description="Swagger UI path")
    redoc_url: str = Field(default=DEFAULT_REDOC_URL, description="ReDoc path")
    redirect_slashes: bool = Field(default=False, description="Redirect between /path and /path/")
    fastapi_kwargs: dict[str, Any] = Field(
        default_factory=dict, description="Merged into the FastAPI(...) call; JSON in the environment"
    )
    uvicorn: UvicornSettings = Field(default_factory=UvicornSettings)
    uvicorn_kwargs: dict[str, Any] = Field(
        default_factory=dict,
        description="Merged into the uvicorn.Config(...) call over the uvicorn section, so it wins on a collision; "
        "JSON in the environment",
    )
    health: HealthSettings = Field(default_factory=HealthSettings)
    middlewares: MiddlewareSettings = Field(default_factory=MiddlewareSettings)

    @model_validator(mode="after")
    def _validate_port(self) -> HttpServerSettings:
        if self.port == 0 and not self.allow_ephemeral_port:
            raise ValueError(
                "server port=0 binds an ephemeral port and a probe or load balancer targeting a fixed port will not "
                "reach it; set allow_ephemeral_port=True if that is intended"
            )
        return self

    def to_config(self, *, version: str = "0.0.0") -> HttpConfig:
        """The ``HttpConfig`` these settings describe; ``version`` is the OpenAPI version, your ``app_version``.

        ``middlewares`` is not part of it: the entrypoint takes that beside the
        config, as ``middlewares=settings.server.middlewares.to_config()``.
        """
        return HttpConfig(
            host=self.host,
            port=self.port,
            graceful_timeout=self.graceful_timeout,
            title=self.title,
            version=version,
            openapi_url=self.openapi_url,
            docs_url=self.docs_url,
            redoc_url=self.redoc_url,
            redirect_slashes=self.redirect_slashes,
            fastapi_kwargs=dict(self.fastapi_kwargs),
            uvicorn_kwargs={**self.uvicorn.to_kwargs(), **self.uvicorn_kwargs},
            health=self.health.to_config(),
        )

class LivenessResponse(BaseModel):
    """Liveness probe body."""

    model_config = ConfigDict(extra="forbid")

    status: Literal["ok"]

@dataclass(slots=True)
class LoggingMiddlewareConfig:
    """Configuration for the request-logging middleware."""

    enabled: bool = True
    ignored_paths: list[str] = field(default_factory=lambda: list(DEFAULT_IGNORED_PATHS))

class LoggingMiddlewareSettings(BaseModel):
    """``LoggingMiddlewareConfig`` as settings; ``ignored_paths`` takes a JSON list."""

    enabled: bool = Field(default=True, description="One structured log line per request")
    ignored_paths: list[str] = Field(
        default_factory=lambda: list(DEFAULT_IGNORED_PATHS), description="Paths left out of request logging"
    )

    def to_config(self) -> LoggingMiddlewareConfig:
        return LoggingMiddlewareConfig(enabled=self.enabled, ignored_paths=list(self.ignored_paths))

@dataclass(slots=True)
class MetricsInstrumentatorConfig:
    """Configuration for ``prometheus-fastapi-instrumentator``."""

    init_kwargs: dict[str, Any] = field(default_factory=dict)
    instrument_kwargs: dict[str, Any] = field(default_factory=dict)
    expose_kwargs: dict[str, Any] = field(default_factory=dict)

    # Passed to ``Instrumentator.add()`` before ``instrument()``: the library's
    # per-request callables (``metrics.request_size()`` and friends, or your
    # own). Any given here REPLACE the default set rather than extend it.
    instrumentations: Sequence[Any] = ()

@dataclass(slots=True)
class MiddlewareConfig:
    """Configuration for the standard platform middleware stack.

    The simple booleans toggle the parameter-less platform middlewares; the
    complex configs carry their own options. ``custom`` lets callers append
    extra ASGI middleware classes (added first so they run last in the stack).
    """

    # Simple toggle-only middlewares.
    unit_scope: bool = True
    """Open one ``UnitScope`` per request (``UnitScopeMiddleware``, outermost).

    Set ``False`` when the framework's own DI integration already owns the
    request scope — dishka's ``setup_dishka`` for instance — so the two never
    open two scopes per request. ``UnitScopeDep`` / ``current_unit_scope()``
    then raise ``LookupError``; resolve through that integration instead.
    """
    context: bool = True
    sentry: bool = True
    processing_time: bool = True

    # Pluggable context propagation: ``None`` = the defaults (structlog +
    # OTel baggage when opentelemetry is installed); pass your own
    # ContextSetter list to replace them.
    context_setters: list[Any] | None = None

    # Complex middlewares.
    logging: LoggingMiddlewareConfig = field(default_factory=LoggingMiddlewareConfig)
    correlation_id: CorrelationIdMiddlewareConfig = field(default_factory=CorrelationIdMiddlewareConfig)
    gzip: GZipMiddlewareConfig = field(default_factory=GZipMiddlewareConfig)
    cors: CORSMiddlewareConfig = field(default_factory=CORSMiddlewareConfig)

    # Custom middlewares: (middleware_class, kwargs).
    custom: list[tuple[type, dict[str, Any]]] = field(default_factory=list)

class MiddlewareSettings(BaseModel):
    """``MiddlewareConfig`` as settings: the toggles and the four configured middlewares.

    ``unit_scope``, ``context_setters`` and ``custom`` are not fields: whether the
    DI integration owns the request scope and which middleware classes to append
    is decided by the code that wires them, so they are the keyword arguments of
    :meth:`to_config`, with the dataclass defaults.
    """

    context: bool = Field(default=True, description="ContextMiddleware: correlation ids, bound and echoed")
    sentry: bool = Field(default=True, description="SentryMiddleware: enrich the Sentry scope from the request")
    processing_time: bool = Field(default=True, description="X-Process-Time response header")
    logging: LoggingMiddlewareSettings = Field(default_factory=LoggingMiddlewareSettings)
    correlation_id: CorrelationIdMiddlewareSettings = Field(default_factory=CorrelationIdMiddlewareSettings)
    gzip: GZipMiddlewareSettings = Field(default_factory=GZipMiddlewareSettings)
    cors: CORSMiddlewareSettings = Field(default_factory=CORSMiddlewareSettings)

    def to_config(
        self,
        *,
        unit_scope: bool = True,
        context_setters: list[Any] | None = None,
        custom: Sequence[tuple[type, dict[str, Any]]] = (),
    ) -> MiddlewareConfig:
        """The ``MiddlewareConfig`` these settings describe; the keywords are the members no environment carries."""
        return MiddlewareConfig(
            unit_scope=unit_scope,
            context=self.context,
            sentry=self.sentry,
            processing_time=self.processing_time,
            context_setters=context_setters,
            logging=self.logging.to_config(),
            correlation_id=self.correlation_id.to_config(),
            gzip=self.gzip.to_config(),
            cors=self.cors.to_config(),
            custom=list(custom),
        )