API reference: servicewright¶

class AppScopeProtocol(Protocol):
    """Protocol for application-level dependency scope.

    The application scope is opened once for the whole process lifetime and
    hosts long-lived singletons (connection pools, clients, ...).
    """

    @overload
    async def get(self, dependency_key: type[T]) -> T: ...

    @overload
    async def get(self, dependency_key: str) -> Any: ...

    async def get(self, dependency_key: type[T] | str) -> T | Any:
        """Retrieve dependency instance by key or type."""
        ...

@dataclass(slots=True)
class AppSpec[TSettings: "BaseServiceSettingsProtocol", TContainer: "DependencyContainerProtocol"]:
    """Complete transport-neutral declarative description of a microservice."""

    service_name: str
    create_container: Callable[[TSettings], TContainer]
    lifecycle: Lifecycle = field(default_factory=Lifecycle)
    observability: ObservabilityManager = field(default_factory=ObservabilityManager)
    health: HealthRegistry = field(default_factory=HealthRegistry)
    warmers: list[AsyncWarmer] = field(default_factory=list)
    warmers_factory: WarmerFactory | None = None
    drain_grace_seconds: float = DEFAULT_DRAIN_GRACE_SECONDS
    """How long each entrypoint gets to finish in-flight work during drain."""
    cleanup_timeout_seconds: float = DEFAULT_CLEANUP_TIMEOUT_SECONDS
    """Budget for each post-drain step (``stop()``, hooks, observability flush)."""

class AsyncWarmer(ABC):
    """Base class for async infrastructure warmers."""

    def __init__(self, *, raise_on_failure: bool = True) -> None:
        """Initialize base warmer behavior."""
        self._raise_on_failure = raise_on_failure

    @property
    def raise_on_failure(self) -> bool:
        """Whether orchestrator should raise when this warmer fails."""
        return self._raise_on_failure

    @property
    def priority(self) -> int:
        """Priority of the warmer (lower value means higher priority).

        Warmers with higher priority (lower values) are executed first.
        Warmers with the same priority are executed in parallel.
        """
        return 0

    @abstractmethod
    async def warmup(self) -> None:
        """Perform infrastructure warmup.

        This method should initialize connection pools, fetch metadata,
        or perform any other operations to prepare the infrastructure
        for high-load production traffic.

        Raises:
            WarmupError: If warmup operation fails and cannot be recovered.
        """
        raise NotImplementedError

class BaseServiceSettingsProtocol(Protocol):
    """Protocol for complete microservice settings.

    Observability sections are named by concern; ``None`` means the concern is
    unconfigured (its sink stays a NullObject).
    """

    @property
    def logging(self) -> LoggingSettingsProtocol | None:
        """Get logging configuration."""
        ...

    @property
    def metrics(self) -> MetricsSettingsProtocol | None:
        """Get metrics configuration."""
        ...

    @property
    def error_tracking(self) -> ErrorTrackingSettingsProtocol | None:
        """Get error-tracking configuration."""
        ...

    @property
    def tracing(self) -> TracingSettingsProtocol | None:
        """Get tracing configuration."""
        ...

    def get_app_version(self) -> str:
        """Return application version string."""
        ...

@dataclass(slots=True)
class BootstrapContext[TSettings: "BaseServiceSettingsProtocol", TContainer: "DependencyContainerProtocol"]:
    """Context available before the application scope is opened."""

    settings: TSettings
    service_name: str
    container: TContainer
    lifecycle: Lifecycle = field(default_factory=Lifecycle)

class ChainRedactor:
    """Applies redactors left to right: ``ChainRedactor(KeyRedactor(), ValueRedactor(m))``.

    Order matters and the conventional order is key-based first: sensitive
    fields are already collapsed to the mask before the (potentially more
    expensive) value masker sees the payload.
    """

    def __init__(self, *redactors: Redactor) -> None:
        self._redactors = redactors

    def __call__(self, data: dict[str, Any]) -> dict[str, Any]:
        """Return ``data`` passed through every redactor in order."""
        for redactor in self._redactors:
            data = redactor(data)
        return data

class CleanupTimeoutError(ServiceWrightError, TimeoutError):
    """Raised when graceful cleanup does not finish within the allotted timeout."""

class ContextSetter(Protocol):
    """Pushes context values into an external system (logging, tracing, ...).

    Implementations receive the full per-unit context dictionary and return a
    cleanup callable that undoes the binding when the unit of work finishes.
    """

    def set(self, context_data: dict[str, Any]) -> Callable[[], None]:
        """Set the context values and return a cleanup callable that resets them."""
        ...

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 DependencyContainerProtocol(Protocol):
    """Protocol for a DI container exposing the two scope tiers."""

    def app_scope(self) -> contextlib.AbstractAsyncContextManager[AppScopeProtocol]:
        """Return async context manager for the process-lifetime application scope."""
        ...

    def unit_scope(
        self, context: Mapping[Any, Any] | None = None
    ) -> contextlib.AbstractAsyncContextManager[UnitScopeProtocol]:
        """Return async context manager for a per-unit-of-work scope.

        ``context`` keys are container-defined: string-keyed payloads and
        type-keyed contexts (e.g. dishka's ``{Request: request}``) both fit.
        """
        ...

class DrainTimeoutError(ServiceWrightError, TimeoutError):
    """Raised when an entrypoint does not drain in-flight work within the grace window."""

@runtime_checkable
class Entrypoint(Protocol):
    """Host-facing protocol implemented by every driver."""

    kind: str
    """Telemetry label only ("http"|"grpc"|"scheduler"|"kafka"|...)."""

    essential: bool
    """If True, this entrypoint's failure/exit stops the whole process.

    A failure also propagates out of ``Host.run`` once cleanup is done, so the
    process exit code distinguishes a crash from a graceful stop.
    """

    async def bind(self, ctx: ServiceContext) -> None:
        """Allocate/subscribe/register. No traffic is accepted yet.

        Raise here if the resource cannot be acquired (a port already in use, a
        missing topic): the Host aborts startup instead of reporting ready.
        """
        ...

    async def serve(self, *, stop: asyncio.Event) -> None:
        """Run until ``stop`` is set, then return **without** shutting down.

        Returning is the signal that the entrypoint is still accepting work and
        is ready to be torn down in order: the Host flips readiness to false
        first (so load balancers stop routing), then calls :meth:`drain`, then
        :meth:`stop`. Closing listeners here instead would make ``drain(grace)``
        inert and would take the readiness endpoint down before the router knows.

        Raise to report a fatal serve-time failure.
        """
        ...

    async def drain(self, grace: float) -> None:
        """Stop intake and let in-flight units finish within ``grace`` seconds."""
        ...

    async def stop(self) -> None:
        """Hard stop / release resources."""
        ...

@dataclass(frozen=True, slots=True)
class ErrorInfo:
    """Normalized view of one failure, ready for rendering.

    Attributes:
        kind: The failure category.
        code: Machine-readable error code.
        detail: Human-readable message (``None`` -> the renderer supplies a
            generic phrase).
        params: Structured, JSON-safe details.
        public: Whether the details may be shown to the client.
        status_override: Explicit HTTP status taking precedence over the kind's
            default (e.g. 422 for request validation).
        headers: Extra response headers (e.g. from an ``HTTPException``).
    """

    kind: ErrorKind
    code: str
    detail: str | None = None
    params: Mapping[str, Any] = field(default_factory=dict)
    public: bool = True
    status_override: int | None = None
    headers: Mapping[str, str] | None = None

    @classmethod
    def from_service_error(cls, exc: ServiceError) -> ErrorInfo:
        """Build the normalized view of a :class:`ServiceError`."""
        return cls(
            kind=exc.kind,
            code=exc.code or INTERNAL_ERROR_CODE,
            detail=exc.detail,
            params=exc.params,
            public=exc.public,
        )

    @property
    def http_status(self) -> int:
        """The HTTP status: the override if set, else the kind's default."""
        return self.status_override if self.status_override is not None else HTTP_STATUS_BY_KIND[self.kind]

class ErrorKind(enum.StrEnum):
    """Transport-neutral failure category (maps to HTTP and gRPC statuses)."""

    INVALID = "invalid"
    UNAUTHENTICATED = "unauthenticated"
    FORBIDDEN = "forbidden"
    NOT_FOUND = "not_found"
    CONFLICT = "conflict"
    PRECONDITION_FAILED = "precondition_failed"
    TOO_MANY_REQUESTS = "too_many_requests"
    DEADLINE_EXCEEDED = "deadline_exceeded"
    UNAVAILABLE = "unavailable"
    NOT_IMPLEMENTED = "not_implemented"
    INTERNAL = "internal"

class ErrorTrackingSettingsProtocol(Protocol):
    """Error-tracking concern: reporting endpoint and sampling."""

    @property
    def dsn(self) -> str | None: ...

    @property
    def environment(self) -> str: ...

    @property
    def traces_sample_rate(self) -> float: ...

    @property
    def profiles_sample_rate(self) -> float: ...

    @property
    def debug(self) -> bool: ...

class HealthCheckerProtocol(Protocol):
    """Protocol for component health checks (DB, Redis, etc.)."""

    async def check(self) -> bool:
        """Check component health. Returns True if healthy."""
        ...

class HealthRegistry:
    """Holds readiness state and named component health checks.

    Liveness reflects only that the process is alive and is independent of the
    registered checks. Readiness requires both the ``ready`` flag (flipped on
    by the Host once serving) and every registered check passing.
    """

    def __init__(self, *, readiness_cache_ttl: float = 0.0) -> None:
        """Initialize the registry.

        Args:
            readiness_cache_ttl: When > 0, readiness results are cached for this
                many seconds to avoid hammering downstream checks.
        """
        self.ready: bool = False
        self._checks: dict[str, HealthCheckerProtocol] = {}
        self._readiness_cache_ttl = readiness_cache_ttl
        self._cached_report: HealthReport | None = None
        self._cached_at: float = 0.0

    def add_check(self, name: str, check: HealthCheckerProtocol) -> None:
        """Register a named component health check.

        Args:
            name: Unique label for the check (e.g. ``"postgres"``).
            check: Object implementing :class:`HealthCheckerProtocol`.
        """
        if name in self._checks:
            raise ValueError(f"Health check '{name}' is already registered")
        self._checks[name] = check
        self._invalidate_cache()

    @property
    def checks(self) -> dict[str, HealthCheckerProtocol]:
        """Return a copy of the registered checks mapping."""
        return dict(self._checks)

    def _invalidate_cache(self) -> None:
        self._cached_report = None
        self._cached_at = 0.0

    async def liveness(self) -> HealthReport:
        """Report process liveness (always healthy while the loop runs)."""
        return HealthReport(healthy=True, checks={})

    async def readiness(self) -> HealthReport:
        """Report readiness: ``ready`` flag AND all checks passing.

        Checks run concurrently; any check raising is treated as a failure.
        """
        if (
            self._readiness_cache_ttl > 0
            and self._cached_report is not None
            and (time.monotonic() - self._cached_at) < self._readiness_cache_ttl
        ):
            return self._cached_report

        results = await self._run_checks()
        healthy = self.ready and all(results.values())
        report = HealthReport(healthy=healthy, checks=results)

        if self._readiness_cache_ttl > 0:
            self._cached_report = report
            self._cached_at = time.monotonic()

        return report

    async def _run_checks(self) -> dict[str, bool]:
        if not self._checks:
            return {}

        names = list(self._checks)
        outcomes = await asyncio.gather(
            *(self._safe_check(name) for name in names),
            return_exceptions=False,
        )
        return dict(zip(names, outcomes, strict=True))

    async def _safe_check(self, name: str) -> bool:
        check = self._checks[name]
        try:
            return bool(await check.check())
        except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
            raise
        except Exception:
            logger.exception("Health check failed", extra={"check": name})
            return False

@dataclass(slots=True, frozen=True)
class HealthReport:
    """Result of a liveness or readiness probe.

    Attributes:
        healthy: Aggregate health flag.
        checks: Per-named-check results (empty for liveness).
    """

    healthy: bool
    checks: dict[str, bool] = field(default_factory=dict)

    @property
    def status(self) -> ProbeStatus:
        """Coarse probe status derived from :attr:`healthy`."""
        return ProbeStatus.from_bool(self.healthy)

class Host[TSettings: "BaseServiceSettingsProtocol", TContainer: "DependencyContainerProtocol"]:
    """Runs an :class:`AppSpec` plus a list of :class:`Entrypoint` drivers.

    Owns the unified lifecycle: Bootstrap -> Warmup -> Ready -> Serve -> Drain
    -> Cleanup. It treats every entrypoint identically and never branches on
    ``kind``.

    The run-loop reports failure by raising, so the process exit code is
    meaningful: an essential entrypoint that dies during ``serve`` propagates its
    exception out of :meth:`run` (after cleanup), and a shutdown step that blows
    past its budget raises :class:`DrainTimeoutError`/:class:`CleanupTimeoutError`
    when nothing else is already propagating.
    """

    def __init__(self, spec: AppSpec[TSettings, TContainer]) -> None:
        self.spec = spec
        self._entrypoints: list[Entrypoint] = []
        self._bound: list[Entrypoint] = []

    def add_entrypoint(self, entrypoint: Entrypoint) -> None:
        """Append an entrypoint (typically from a plugin's ``on_register``)."""
        self._entrypoints.append(entrypoint)

    def bootstrap(self, settings: TSettings) -> BootstrapContext[TSettings, TContainer]:
        """Build the container (the application scope is not yet entered)."""
        return BootstrapContext(
            settings=settings,
            service_name=self.spec.service_name,
            container=self.spec.create_container(settings),
            lifecycle=self.spec.lifecycle,
        )

    async def run(
        self,
        settings: TSettings,
        entrypoints: Iterable[Entrypoint] = (),
        *,
        plugins: Iterable[Plugin] = (),
        stop: asyncio.Event | None = None,
    ) -> None:
        """Run the full lifecycle, blocking until ``stop`` is set.

        Args:
            settings: Service settings.
            entrypoints: Drivers to run.
            plugins: Plugins applied via ``on_register`` before the run-loop.
            stop: Externally supplied stop event. When provided, OS signal
                handlers are NOT installed (the embedding/test path).

        Raises:
            Exception: Whatever an essential entrypoint raised while serving, or
                whatever startup raised — after cleanup has run.
            ServiceWrightError: If a shutdown step exceeded its budget and no
                other exception is propagating.
        """
        self._entrypoints = list(entrypoints)
        self._bound = []
        for plugin in plugins:
            plugin.on_register(self.spec, self)

        self.spec.observability.configure(settings, service_name=self.spec.service_name)
        uninstall_signal_handlers: Callable[[], None] | None = None
        timeouts: list[ServiceWrightError] = []
        try:
            bootstrap_ctx = self.bootstrap(settings)

            if stop is None:
                stop = asyncio.Event()
                uninstall_signal_handlers = install_signal_handlers(stop)

            await self._run_with_app_scope(bootstrap_ctx, stop)
        finally:
            if uninstall_signal_handlers is not None:
                uninstall_signal_handlers()
            timeouts.extend(await self._final_cleanup())

        # Only reachable when nothing else is propagating: a shutdown that blew
        # its budget must not mask the failure that caused the shutdown.
        if timeouts:
            raise timeouts[0]

    async def _final_cleanup(self) -> list[ServiceWrightError]:
        """Flush observability and run post-shutdown hooks, both time-boxed."""
        budget = self.spec.cleanup_timeout_seconds
        timeouts: list[ServiceWrightError] = []
        # Sink shutdown flushes (traces, sentry events) and joins the metrics
        # server thread — off the loop so it can never block the last steps.
        timeouts.extend(
            await self._shutdown_step(
                asyncio.to_thread(self.spec.observability.shutdown),
                budget=budget,
                error=CleanupTimeoutError,
                phase="observability shutdown",
            )
        )
        timeouts.extend(
            await self._shutdown_step(
                self.spec.lifecycle.run_post_shutdown_hooks(None),
                budget=budget,
                error=CleanupTimeoutError,
                phase="post-shutdown hooks",
            )
        )
        return timeouts

    async def _run_with_app_scope(
        self,
        bootstrap_ctx: BootstrapContext[TSettings, TContainer],
        stop: asyncio.Event,
    ) -> None:
        async with bootstrap_ctx.container.app_scope() as app_scope:
            service_ctx: ServiceContext[TSettings, TContainer] = ServiceContext(
                bootstrap=bootstrap_ctx,
                app_scope=app_scope,
                health=self.spec.health,
                observability=self.spec.observability,
            )
            timeouts: list[ServiceWrightError] = []
            try:
                if await self._startup(service_ctx, self._entrypoints, stop):
                    await self._serve(self._entrypoints, stop)
            finally:
                timeouts = await self._shutdown_in_scope(service_ctx)

            if timeouts:
                raise timeouts[0]

    async def _startup(
        self,
        service_ctx: ServiceContext[TSettings, TContainer],
        entrypoints: Sequence[Entrypoint],
        stop: asyncio.Event,
    ) -> bool:
        """Warm up, bind every entrypoint and flip readiness.

        Returns:
            ``True`` when the service is fully started and should serve.
            ``False`` when ``stop`` arrived first — startup is abandoned at the
            next phase boundary rather than binding ports and flipping readiness
            on a process that has already been asked to terminate.
        """
        if stop.is_set():
            return self._abandon_startup("before warmup")

        await self._collect_and_warmup(service_ctx, stop)
        if stop.is_set():
            return self._abandon_startup("during warmup")

        await self.spec.lifecycle.run_pre_start_hooks(service_ctx.app_scope)
        for entrypoint in entrypoints:
            if stop.is_set():
                return self._abandon_startup("during bind")
            # Recorded before the await: a bind that fails halfway through has
            # already allocated something, so it must still be torn down.
            self._bound.append(entrypoint)
            await entrypoint.bind(service_ctx)

        if stop.is_set():
            return self._abandon_startup("after bind")

        self.spec.health.ready = True
        await self.spec.lifecycle.run_post_start_hooks(service_ctx.app_scope)
        logger.info("Service ready", extra={"service": self.spec.service_name})
        return True

    def _abandon_startup(self, phase: str) -> bool:
        logger.info(
            "Stop requested during startup; skipping serve",
            extra={"service": self.spec.service_name, "phase": phase},
        )
        return False

    async def _collect_and_warmup(
        self,
        service_ctx: ServiceContext[TSettings, TContainer],
        stop: asyncio.Event,
    ) -> None:
        """Prime every warmer, abandoning the wait as soon as ``stop`` is set."""
        warmers = await collect_warmers(
            base_warmers=self.spec.warmers,
            warmers_factory=self.spec.warmers_factory,
            app_ctx=service_ctx,
        )
        warmup = asyncio.ensure_future(perform_warmup(self.spec.service_name, list(warmers)))
        waiter = asyncio.ensure_future(stop.wait())
        try:
            await asyncio.wait({warmup, waiter}, return_when=asyncio.FIRST_COMPLETED)
        finally:
            waiter.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await waiter

        if warmup.done():
            await warmup  # Propagate a warmup failure to the caller of run().
            return

        warmup.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await warmup
        logger.warning(
            "Warmup abandoned: stop requested",
            extra={"service": self.spec.service_name},
        )

    async def _serve(self, entrypoints: Sequence[Entrypoint], stop: asyncio.Event) -> None:
        """Serve every entrypoint until ``stop``; re-raise an essential failure."""
        if not entrypoints:
            await stop.wait()
            return

        async def runner(entrypoint: Entrypoint) -> None:
            try:
                await entrypoint.serve(stop=stop)
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception(
                    "Entrypoint serve failed",
                    extra={"service": self.spec.service_name, "kind": entrypoint.kind},
                )
                if entrypoint.essential:
                    stop.set()
                    raise
                return
            if entrypoint.essential and not stop.is_set():
                logger.info(
                    "Essential entrypoint exited; stopping service",
                    extra={"service": self.spec.service_name, "kind": entrypoint.kind},
                )
                stop.set()

        failure: BaseException | None = None
        try:
            async with asyncio.TaskGroup() as group:
                for entrypoint in entrypoints:
                    group.create_task(runner(entrypoint))
        except* Exception as eg:
            logger.exception(
                "Serve loop aborted by entrypoint failure",
                extra={"service": self.spec.service_name, "errors": len(eg.exceptions)},
            )
            failure = _sole_cause(eg)

        # Only essential failures reach here (runner() swallows the rest), and an
        # essential failure must reach the caller: a process that dies mid-serve
        # may not exit 0, or every exit-code-based supervisor reads it as success.
        if failure is not None:
            raise failure

    async def _shutdown_in_scope(
        self,
        service_ctx: ServiceContext[TSettings, TContainer],
    ) -> list[ServiceWrightError]:
        """Drain, stop and run pre-shutdown hooks; return any budget overruns."""
        # Stop routing first (k8s/LB) before we stop accepting work.
        self.spec.health.ready = False

        grace = self.spec.drain_grace_seconds
        budget = self.spec.cleanup_timeout_seconds
        timeouts: list[ServiceWrightError] = []

        for entrypoint in reversed(self._bound):
            timeouts.extend(
                await self._shutdown_step(
                    entrypoint.drain(grace),
                    budget=grace + DEFAULT_DRAIN_TIMEOUT_BUFFER_SECONDS,
                    error=DrainTimeoutError,
                    phase="drain",
                    kind=entrypoint.kind,
                )
            )

        for entrypoint in reversed(self._bound):
            timeouts.extend(
                await self._shutdown_step(
                    entrypoint.stop(),
                    budget=budget,
                    error=CleanupTimeoutError,
                    phase="stop",
                    kind=entrypoint.kind,
                )
            )

        timeouts.extend(
            await self._shutdown_step(
                self.spec.lifecycle.run_pre_shutdown_hooks(service_ctx.app_scope),
                budget=budget,
                error=CleanupTimeoutError,
                phase="pre-shutdown hooks",
            )
        )

        logger.info("Service shutdown complete", extra={"service": self.spec.service_name})
        return timeouts

    async def _shutdown_step(
        self,
        step: Coroutine[object, object, None] | Awaitable[None],
        *,
        budget: float,
        error: type[ServiceWrightError],
        phase: str,
        kind: str | None = None,
    ) -> list[ServiceWrightError]:
        """Await one shutdown step under a budget, never letting it abort the rest.

        A failing step is logged and skipped so the remaining entrypoints still
        get their turn; a step that exceeds ``budget`` is returned to the caller,
        which raises it once every other step has had its chance.
        """
        extra = {"service": self.spec.service_name, "phase": phase, "kind": kind, "budget": budget}
        try:
            await asyncio.wait_for(step, timeout=budget)
        except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
            raise
        except TimeoutError as exc:
            timeout_error = error(f"{phase} did not finish within {budget}s")
            timeout_error.__cause__ = exc
            logger.warning("Shutdown step timed out", extra=extra, exc_info=timeout_error)
            return [timeout_error]
        except Exception:
            logger.exception("Shutdown step failed", extra=extra)
        return []