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."
            )

@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 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

@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

@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 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))

@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)

@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 OtelBaggageSetter(ContextSetter):
    """Put request/user/trace identifiers into OpenTelemetry Baggage.

    Baggage rides the W3C ``baggage`` header, so OTel-instrumented clients
    (httpx, grpc, kafka) propagate the values to downstream services with no
    custom code. Filter propagators on clients calling third parties if these
    identifiers must not leave your system.

    Raises:
        ImportError: If ``opentelemetry-api`` is not installed.
    """

    def __init__(self) -> None:
        if not OTEL_AVAILABLE:
            raise ImportError(_OTEL_HINT)

    def set(self, context_data: dict[str, Any]) -> Callable[[], None]:
        """Attach one OTel context carrying the values; return the detacher."""
        ctx = _otel_context.get_current()
        dirty = False
        for key in _CONTEXT_KEYS:
            if value := context_data.get(key):
                ctx = _otel_baggage.set_baggage(key, str(value), context=ctx)
                dirty = True

        token = _otel_context.attach(ctx) if dirty else None
        return OtelBaggageRemover(token)

class ProblemDetails(BaseModel):
    """RFC 9457 Problem Details document (``application/problem+json``).

    The shape produced by the default
    :class:`~servicewright.core.errors.ProblemDetailsRenderer`; extra members
    are allowed per the RFC (custom renderers may add their own extensions).
    """

    model_config = ConfigDict(extra="allow")

    type: str = "about:blank"
    title: str
    status: int
    detail: str | None = None
    code: str | None = None
    params: dict[str, Any] | None = Field(default=None)

class ReadinessResponse(BaseModel):
    """Readiness probe body."""

    model_config = ConfigDict(extra="forbid")

    status: Literal["ok", "unhealthy"]

class StructlogSetter(ContextSetter):
    """Bind request/user/trace identifiers into ``structlog`` contextvars.

    Raises:
        ImportError: If ``structlog`` is not installed.
    """

    def __init__(self) -> None:
        if not STRUCTLOG_AVAILABLE:
            raise ImportError(_STRUCTLOG_HINT)

    def set(self, context_data: dict[str, Any]) -> Callable[[], None]:
        """Bind the contextvars and return a remover."""
        bind_data: dict[str, Any] = {}
        bound_keys: list[str] = []
        for key in _CONTEXT_KEYS:
            if value := context_data.get(key):
                bind_data[key] = value
                bound_keys.append(key)

        if bind_data:
            structlog.contextvars.bind_contextvars(**bind_data)

        return StructlogRemover(bound_keys)

class UnitScopeMiddleware:
    """Open one ``UnitScope`` per request and expose it three ways.

    Installed by the :class:`FastApiEntrypoint` as the outermost wrapper around
    the handler unless ``MiddlewareConfig.unit_scope`` is off. The scope
    carries the ``Request`` as its ``context`` and stays open until the
    response is fully delivered.

    This is deliberately a raw ASGI middleware rather than a
    ``BaseHTTPMiddleware``: the latter hands control back as soon as the
    response *starts*, so the scope — and with it every REQUEST-scoped
    dependency, e.g. a database session — would be finalized while a streaming
    body is still being produced and before ``BackgroundTask``s run, truncating
    responses that the client already received a 200 for. Awaiting the inner app
    to completion keeps the scope alive for the whole exchange, which is also
    what the Litestar adapter does.
    """

    def __init__(self, app: ASGIApp, container: DependencyContainerProtocol) -> None:
        self._app = app
        self._container = container

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Wrap the request in a fresh unit scope bound to request.state + a contextvar."""
        if scope["type"] != "http":
            await self._app(scope, receive, send)
            return

        request = Request(scope)
        async with self._container.unit_scope({"request": request}) as unit_scope:
            request.state.unit_scope = unit_scope
            token = _current_unit_scope.set(unit_scope)
            try:
                await self._app(scope, receive, send)
            finally:
                _current_unit_scope.reset(token)