spakky-actuator

Actuator 상태/정보 계약 — Health, Readiness, Liveness, Info

결과 계약

Transport-neutral actuator result contracts.

HealthStatus

Bases: StrEnum

Aggregate or component health status.

ActuatorEndpoint

Bases: StrEnum

Transport-neutral actuator endpoint kinds.

ComponentHealthResult

Health result for one actuator component.

healthy(name, *, required=True, details=None) classmethod

Create a healthy component result.

Source code in core/spakky-actuator/src/spakky/actuator/result.py

@classmethod
def healthy(
    cls,
    name: str,
    *,
    required: bool = True,
    details: Mapping[str, object] | None = None,
) -> Self:
    """Create a healthy component result."""
    return cls(
        name=name,
        status=HealthStatus.HEALTHY,
        required=required,
        details=_sorted_mapping(details),
    )

unhealthy(name, *, required=True, details=None) classmethod

Create an unhealthy component result.

Source code in core/spakky-actuator/src/spakky/actuator/result.py

@classmethod
def unhealthy(
    cls,
    name: str,
    *,
    required: bool = True,
    details: Mapping[str, object] | None = None,
) -> Self:
    """Create an unhealthy component result."""
    return cls(
        name=name,
        status=HealthStatus.UNHEALTHY,
        required=required,
        details=_sorted_mapping(details),
    )

ActuatorHealthResult

Aggregate actuator health result.

healthy_baseline(endpoint) classmethod

Create deterministic healthy baseline result for no probes.

Source code in core/spakky-actuator/src/spakky/actuator/result.py

@classmethod
def healthy_baseline(cls, endpoint: ActuatorEndpoint) -> Self:
    """Create deterministic healthy baseline result for no probes."""
    return cls(endpoint=endpoint, status=HealthStatus.HEALTHY)

from_components(endpoint, components) classmethod

Create aggregate result from component results.

Source code in core/spakky-actuator/src/spakky/actuator/result.py

@classmethod
def from_components(
    cls,
    endpoint: ActuatorEndpoint,
    components: tuple[ComponentHealthResult, ...],
) -> Self:
    """Create aggregate result from component results."""
    ordered_components = tuple(sorted(components, key=lambda item: item.name))
    if not ordered_components:
        return cls.healthy_baseline(endpoint)
    status = HealthStatus.HEALTHY
    for component in ordered_components:
        if component.required and component.status is HealthStatus.UNHEALTHY:
            status = HealthStatus.UNHEALTHY
            break
    return cls(
        endpoint=endpoint,
        status=status,
        components=ordered_components,
    )

ActuatorInfoResult

Transport-neutral actuator info result.

from_mapping(info) classmethod

Create deterministic info result from a mapping.

Source code in core/spakky-actuator/src/spakky/actuator/result.py

@classmethod
def from_mapping(cls, info: Mapping[str, object]) -> Self:
    """Create deterministic info result from a mapping."""
    return cls(info=_sorted_mapping(info))

Service

Actuator aggregation service.

ActuatorAggregationService(registry, config=None)

Aggregate health probes and info contributors into actuator results.

Initialize the aggregation service.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

def __init__(
    self,
    registry: ActuatorExtensionRegistry,
    config: ActuatorConfig | None = None,
) -> None:
    """Initialize the aggregation service."""
    self._registry = registry
    self._config = ActuatorConfig() if config is None else config

evaluate_health()

Evaluate the health endpoint synchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

def evaluate_health(self) -> ActuatorHealthResult:
    """Evaluate the health endpoint synchronously."""
    return self._evaluate_sync(ActuatorEndpoint.HEALTH)

evaluate_readiness()

Evaluate the readiness endpoint synchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

def evaluate_readiness(self) -> ActuatorHealthResult:
    """Evaluate the readiness endpoint synchronously."""
    return self._evaluate_sync(ActuatorEndpoint.READINESS)

evaluate_liveness()

Evaluate the liveness endpoint synchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

def evaluate_liveness(self) -> ActuatorHealthResult:
    """Evaluate the liveness endpoint synchronously."""
    return self._evaluate_sync(ActuatorEndpoint.LIVENESS)

evaluate_health_async() async

Evaluate the health endpoint asynchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

async def evaluate_health_async(self) -> ActuatorHealthResult:
    """Evaluate the health endpoint asynchronously."""
    return await self._evaluate_async(ActuatorEndpoint.HEALTH)

evaluate_readiness_async() async

Evaluate the readiness endpoint asynchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

async def evaluate_readiness_async(self) -> ActuatorHealthResult:
    """Evaluate the readiness endpoint asynchronously."""
    return await self._evaluate_async(ActuatorEndpoint.READINESS)

evaluate_liveness_async() async

Evaluate the liveness endpoint asynchronously.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

async def evaluate_liveness_async(self) -> ActuatorHealthResult:
    """Evaluate the liveness endpoint asynchronously."""
    return await self._evaluate_async(ActuatorEndpoint.LIVENESS)

evaluate_info()

Evaluate synchronous info contributors.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

def evaluate_info(self) -> ActuatorInfoResult:
    """Evaluate synchronous info contributors."""
    if self._matching_async_info_contributors():
        raise CannotEvaluateAsyncExtensionSynchronouslyError()
    return ActuatorInfoResult.from_mapping(
        self._merge_info(self._registry.info_contributors())
    )

evaluate_info_async() async

Evaluate sync and async info contributors.

Source code in core/spakky-actuator/src/spakky/actuator/service.py

async def evaluate_info_async(self) -> ActuatorInfoResult:
    """Evaluate sync and async info contributors."""
    info = self._merge_info(self._registry.info_contributors())
    for contributor in self._registry.async_info_contributors():
        info.update(await contributor.contribute_info_async())
    return ActuatorInfoResult.from_mapping(info)

Registry

Registry for DI-managed actuator extensions.

ActuatorExtensionRegistry()

In-memory registry populated from DI-managed actuator extensions.

Initialize an empty actuator extension registry.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def __init__(self) -> None:
    """Initialize an empty actuator extension registry."""
    self._health_probes = []
    self._async_health_probes = []
    self._info_contributors = []
    self._async_info_contributors = []

register_health_probe(probe)

Register a synchronous health probe.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def register_health_probe(self, probe: AbstractHealthProbe) -> None:
    """Register a synchronous health probe."""
    self._health_probes.append(probe)

register_async_health_probe(probe)

Register an asynchronous health probe.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def register_async_health_probe(self, probe: AbstractAsyncHealthProbe) -> None:
    """Register an asynchronous health probe."""
    self._async_health_probes.append(probe)

register_info_contributor(contributor)

Register a synchronous info contributor.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def register_info_contributor(self, contributor: IInfoContributor) -> None:
    """Register a synchronous info contributor."""
    self._info_contributors.append(contributor)

register_async_info_contributor(contributor)

Register an asynchronous info contributor.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def register_async_info_contributor(
    self, contributor: IAsyncInfoContributor
) -> None:
    """Register an asynchronous info contributor."""
    self._async_info_contributors.append(contributor)

health_probes()

Return registered synchronous health probes sorted by name.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def health_probes(self) -> tuple[AbstractHealthProbe, ...]:
    """Return registered synchronous health probes sorted by name."""
    return tuple(sorted(self._health_probes, key=lambda probe: probe.name))

async_health_probes()

Return registered asynchronous health probes sorted by name.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def async_health_probes(self) -> tuple[AbstractAsyncHealthProbe, ...]:
    """Return registered asynchronous health probes sorted by name."""
    return tuple(sorted(self._async_health_probes, key=lambda probe: probe.name))

info_contributors()

Return registered synchronous info contributors sorted by name.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def info_contributors(self) -> tuple[IInfoContributor, ...]:
    """Return registered synchronous info contributors sorted by name."""
    return tuple(
        sorted(self._info_contributors, key=lambda contributor: contributor.name)
    )

async_info_contributors()

Return registered asynchronous info contributors sorted by name.

Source code in core/spakky-actuator/src/spakky/actuator/registry.py

def async_info_contributors(self) -> tuple[IAsyncInfoContributor, ...]:
    """Return registered asynchronous info contributors sorted by name."""
    return tuple(
        sorted(
            self._async_info_contributors,
            key=lambda contributor: contributor.name,
        )
    )

확장 인터페이스

Health probe extension points.

AbstractHealthProbe

Bases: ABC

Synchronous health probe extension point.

name abstractmethod property

Stable probe name included in component results.

required property

Whether an unhealthy result should fail the aggregate status.

endpoints property

Actuator endpoints evaluated by this probe.

check() abstractmethod

Evaluate this probe.

Source code in core/spakky-actuator/src/spakky/actuator/interfaces/probe.py

@abstractmethod
def check(self) -> ComponentHealthResult:
    """Evaluate this probe."""
    ...

AbstractAsyncHealthProbe

Bases: ABC

Asynchronous health probe extension point.

name abstractmethod property

Stable probe name included in component results.

required property

Whether an unhealthy result should fail the aggregate status.

endpoints property

Actuator endpoints evaluated by this probe.

check_async() abstractmethod async

Evaluate this probe asynchronously.

Source code in core/spakky-actuator/src/spakky/actuator/interfaces/probe.py

@abstractmethod
async def check_async(self) -> ComponentHealthResult:
    """Evaluate this probe asynchronously."""
    ...

Info contributor extension points.

IInfoContributor

Bases: ABC

Synchronous contributor for actuator info output.

name abstractmethod property

Stable contributor name used for deterministic merge ordering.

contribute_info() abstractmethod

Return info entries contributed by this extension.

Source code in core/spakky-actuator/src/spakky/actuator/interfaces/contributor.py

@abstractmethod
def contribute_info(self) -> Mapping[str, object]:
    """Return info entries contributed by this extension."""
    ...

IAsyncInfoContributor

Bases: ABC

Asynchronous contributor for actuator info output.

name abstractmethod property

Stable contributor name used for deterministic merge ordering.

contribute_info_async() abstractmethod async

Return info entries contributed by this extension.

Source code in core/spakky-actuator/src/spakky/actuator/interfaces/contributor.py

@abstractmethod
async def contribute_info_async(self) -> Mapping[str, object]:
    """Return info entries contributed by this extension."""
    ...

기본 Contributor

Built-in actuator info contributors.

StartupReportInfoContributor(report_provider)

Bases: IInfoContributor

Expose startup diagnostics through actuator info.

Source code in core/spakky-actuator/src/spakky/actuator/contributors.py

def __init__(self, report_provider: Callable[[], StartupReport]) -> None:
    self._report_provider = report_provider

설정

Actuator configuration contract.

ActuatorConfig(include_details=True)

Configuration for actuator aggregation behavior.

Initialize actuator configuration.

Parameters:

Name	Type	Description	Default
include_details	bool	Whether component details should be exposed.	True

Source code in core/spakky-actuator/src/spakky/actuator/config.py

def __init__(self, include_details: bool = True) -> None:
    """Initialize actuator configuration.

    Args:
        include_details: Whether component details should be exposed.
    """
    self.include_details = include_details

후처리기

Actuator extension registration post-processor.

ActuatorExtensionPostProcessor(registry)

Bases: IPostProcessor

Collect DI-managed actuator extension pods into the registry.

Initialize with the shared actuator extension registry.

Source code in core/spakky-actuator/src/spakky/actuator/post_processor.py

def __init__(self, registry: ActuatorExtensionRegistry) -> None:
    """Initialize with the shared actuator extension registry."""
    self._registry = registry

post_process(pod)

Register actuator extension pods and return them unmodified.

Source code in core/spakky-actuator/src/spakky/actuator/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Register actuator extension pods and return them unmodified."""
    if isinstance(pod, AbstractHealthProbe):
        self._registry.register_health_probe(pod)
    if isinstance(pod, AbstractAsyncHealthProbe):
        self._registry.register_async_health_probe(pod)
    if isinstance(pod, IInfoContributor):
        self._registry.register_info_contributor(pod)
    if isinstance(pod, IAsyncInfoContributor):
        self._registry.register_async_info_contributor(pod)
    return pod

플러그인 진입점

Plugin initialization entry point.

initialize(app)

Initialize the spakky-actuator plugin.

Parameters:

Name	Type	Description	Default
app	SpakkyApplication	The SpakkyApplication instance.	required

Source code in core/spakky-actuator/src/spakky/actuator/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the spakky-actuator plugin.

    Args:
        app: The SpakkyApplication instance.
    """
    app.add(ActuatorConfig)
    app.add(ActuatorExtensionRegistry)
    app.add(_startup_report_info_contributor(app))
    app.add(ActuatorExtensionPostProcessor)
    app.add(ActuatorAggregationService)

에러

Error classes for the spakky-actuator package.

AbstractSpakkyActuatorError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for actuator-related errors.

CannotEvaluateAsyncExtensionSynchronouslyError

Bases: AbstractSpakkyActuatorError

Raised when a sync evaluation sees an async-only extension.