Configuration Reference¶

@runtime_checkable
class HealthIndicator(Protocol):
    """A component contributing one entry to ``/actuator/health``.

    Attributes:
        name: Stable key shown under ``components`` in the health payload.

    The ``check`` method may be sync or async and may return either a mapping
    (``{"status": "UP", ...}``) or a plain truthy/falsy value.
    """

    name: str

    def check(self) -> Dict[str, Any] | bool:
        """Probe the dependency. Return a mapping or a truthy value for UP."""
        ...

@runtime_checkable
class InfoContributor(Protocol):
    """A component contributing key/values to ``/actuator/info``."""

    def contribute(self) -> Dict[str, Any]:
        """Return a mapping merged into the /info payload."""
        ...

@configured(target="self", prefix="actuator", mapping="tree")
@dataclass
class ActuatorSettings:
    """Type-safe actuator settings, populated from the ``actuator`` prefix.

    Mirrors ``FastApiSettings`` in pico-fastapi.

    Attributes:
        enabled: Master switch for the actuator endpoints.
        show_components: Include per-indicator detail in ``/health``.
        check_timeout_seconds: Per-indicator time budget; a slower ``check()``
            reports ``DOWN`` with a timeout error instead of hanging the endpoint.
        info: Static key/values exposed by ``/info`` (build, git sha, ...).

    Example:
        .. code-block:: yaml

            # application.yaml
            actuator:
              enabled: true
              show_components: true
              check_timeout_seconds: 5.0
              info:
                app: my-service
                build: "2026.06"
    """

    enabled: bool = True
    show_components: bool = True
    check_timeout_seconds: float = 5.0
    info: Dict[str, Any] = field(default_factory=dict)

@controller(prefix="/actuator", tags=["actuator"])
class ActuatorController:
    def __init__(
        self,
        settings: ActuatorSettings,
        indicators: List[HealthIndicator],
        contributors: List[InfoContributor],
        container: PicoContainer,
    ):
        self.settings = settings
        self.indicators = indicators
        self.contributors = contributors
        self.container = container

    async def _gather(self):
        return await gather(self.indicators, timeout=self.settings.check_timeout_seconds)

    @get("/health")
    async def health(self):
        """Full health: overall status plus per-component detail."""
        if not self.settings.enabled:
            return _DISABLED
        overall, components = await self._gather()
        body = {"status": overall}
        if self.settings.show_components:
            body["components"] = components
        # (content, status) tuple — pico-fastapi normalizes this to a JSONResponse.
        return body, 200 if overall == UP else 503

    @get("/health/live")
    async def liveness(self):
        """Liveness: is the process responding at all? Cheap, touches no deps."""
        if not self.settings.enabled:
            return _DISABLED
        return {"status": UP}

    @get("/health/ready")
    async def readiness(self):
        """Readiness: are dependencies healthy? Aggregate of all indicators."""
        if not self.settings.enabled:
            return _DISABLED
        overall, _ = await self._gather()
        return {"status": overall}, 200 if overall == UP else 503

    @get("/info")
    async def info(self):
        """Static config info merged with dynamic contributors."""
        if not self.settings.enabled:
            return _DISABLED
        data = dict(self.settings.info)
        for c in self.contributors:
            data.update(c.contribute())
        return data

    @post("/refresh")
    async def refresh(self):
        """Re-read config sources; report the changed top-level prefixes.

        Mirror of Spring Cloud's ``POST /actuator/refresh``: components
        subscribed to ``ConfigChanged`` re-read their config.
        """
        if not self.settings.enabled:
            return _DISABLED
        return {"changed": sorted(self.container.refresh_config())}

    @get("/metrics")
    async def metrics(self, request: Request):
        """Prometheus/OpenMetrics exposition of the default registry.

        Honors the ``Accept`` header (``application/openmetrics-text`` or
        Prometheus text 0.0.4). Needs the ``metrics`` extra; answers 501
        when prometheus-client is not installed.
        """
        if not self.settings.enabled:
            return _DISABLED
        try:
            from prometheus_client import REGISTRY
            from prometheus_client.exposition import choose_encoder
        except ImportError:
            return {"detail": "prometheus-client not installed; pip install pico-actuator[metrics]"}, 501
        encoder, content_type = choose_encoder(request.headers.get("accept", ""))
        return Response(content=encoder(REGISTRY), media_type=content_type)

class PicoActuatorError(Exception):
    """Base exception for all pico-actuator errors.

    Catch this at startup boundaries to handle any pico-actuator failure
    without matching individual subclasses.
    """