Health
- Start here: Health Checks guide
- FastAPI integration:
health_routerfor liveness, readiness, and health endpoints.
grelmicro.health
Health Checks.
HealthCheckFunc
HealthCheckFunc = SyncHealthCheckFunc | AsyncHealthCheckFunc
Any callable acceptable as a health check.
Returns:
- None: healthy, no details.
- HealthDetails: healthy, with a details dict.
Raises:
- HealthError: unhealthy. The message surfaces in the response.
- Any other exception: unhealthy with a generic message. The
traceback is logged server-side.
HealthDetails
HealthDetails = dict[str, JSONEncodable]
Per-check details payload. JSON-serializable dict keyed by string.
CheckResult
Bases: TypedDict
Result of a single health check.
critical
instance-attribute
critical: bool
error
instance-attribute
error: str | None
HealthError
HealthError(
message: str, *, details: HealthDetails | None = None
)
Bases: GrelmicroError
Signal a check failure. The message is exposed in the response.
Pass details to include a diagnostic payload alongside the
error, visible under details on the check entry in
/healthz (subject to show_details).
Initialize with a message and optional details dict.
details
instance-attribute
details = details
HealthChecks
HealthChecks(
*,
name: str = "default",
timeout: PositiveFloat | None = None,
cache_ttl: NonNegativeFloat | None = None,
env_prefix: str | None = None,
env_load: bool | None = None,
auto_health: bool = False,
)
Bases: Reconfigurable[HealthChecksConfig]
Manages health checks and runs them concurrently.
Checks are plain async functions. Register them with the
:meth:check decorator or the :meth:add method. All registered
checks are executed in parallel via an asyncio.TaskGroup. Each
check has its own timeout (falling back to the default)
and its own cached result. Concurrent requests for the same check
share a single execution via an asyncio.Event.
Supports live reconfiguration via
reconfigure(new_config).
A swap takes effect on the next :meth:run. In-flight rounds
keep the cache_ttl they started with. The new default
timeout applies to checks registered after the swap.
Existing checks keep the timeout they were registered with.
Re-register a check to pick up the new default. See
Live reconfiguration.
Initialize the health checks.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Registration name. Multiple
TYPE:
|
timeout
|
Default per-check timeout in seconds. Checks that
exceed this duration are reported as Default: 5.0. When unset and env reads are enabled (see
TYPE:
|
cache_ttl
|
Per-check cache TTL in seconds. Set to 0 to disable. Default: 1.0. When unset and env reads are enabled (see
TYPE:
|
env_prefix
|
Override the auto-derived environment variable prefix. Default:
TYPE:
|
env_load
|
Whether to read environment variables. When None (the default), follow the process-wide
TYPE:
|
auto_health
|
Register a
TYPE:
|
kind
class-attribute
kind: str = 'health'
name
property
name: str
Return the registration name.
from_config
classmethod
from_config(
config: HealthChecksConfig,
*,
name: str = "default",
auto_health: bool = False,
) -> Self
Construct a HealthChecks from a pre-built HealthChecksConfig.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The pre-built health checks configuration. Use this path when the configuration is assembled at
startup from a settings tree (for example YAML, Vault,
or a
TYPE:
|
name
|
Registration name. Defaults to
TYPE:
|
auto_health
|
Register a
TYPE:
|
add
add(
name: str,
func: HealthCheckFunc,
*,
critical: bool = True,
timeout: PositiveFloat | None = None,
) -> None
Register a health check function.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Unique name identifying this check.
TYPE:
|
func
|
Async function: returns
TYPE:
|
critical
|
Whether this check affects the aggregate status and HTTP response code. Critical failures flip the aggregate to
TYPE:
|
timeout
|
Per-check timeout override. Falls back to the default when omitted.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
check
check(
name: str,
*,
critical: bool = True,
timeout: PositiveFloat | None = None,
) -> Callable[[HealthCheckFunc], HealthCheckFunc]
Decorate an async function to register it as a health check.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
Unique name identifying this check.
TYPE:
|
critical
|
Whether this check affects the aggregate status.
TYPE:
|
timeout
|
Per-check timeout override.
TYPE:
|
Example
@registry.check("database") ... async def check_db() -> dict | None: ... return None
add_provider
add_provider(
provider: Provider,
*,
name: str | None = None,
critical: bool = True,
timeout: PositiveFloat | None = None,
) -> None
Register a provider's built-in readiness check as provider:{name}.
| PARAMETER | DESCRIPTION |
|---|---|
provider
|
The provider whose built-in readiness check to register.
TYPE:
|
name
|
Check name suffix. Defaults to the provider's
TYPE:
|
critical
|
Whether the check affects
TYPE:
|
timeout
|
Per-check timeout override. Falls back to the default.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If the provider ships no readiness check, or the resulting name is already registered. |
run
async
run(
*,
critical_only: bool = False,
exclude: Iterable[str] | None = None,
) -> HealthReport
Run the selected checks concurrently and aggregate.
Each check runs with its own timeout. Results are cached per
check for cache_ttl seconds. Concurrent calls for the
same check coalesce via single-flight.
| PARAMETER | DESCRIPTION |
|---|---|
critical_only
|
If True, only run critical checks.
TYPE:
|
exclude
|
Check names to skip.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
HealthReport
|
A HealthReport with the aggregate status and per-check |
HealthReport
|
results. |
HealthChecksConfig
Bases: BaseModel
Health Checks Config.
timeout
class-attribute
instance-attribute
timeout: PositiveFloat = 5.0
Default per-check timeout in seconds. Checks that exceed this duration are reported as error. Can be overridden per check on registration.
cache_ttl
class-attribute
instance-attribute
cache_ttl: NonNegativeFloat = 1.0
Per-check cache TTL in seconds. Each check's last result is reused until it is older than cache_ttl. Concurrent calls coalesce via single-flight. Set to 0 to disable caching.
HealthReport
Bases: TypedDict
Aggregated health report across all registered checks.
HealthStatus
Bases: StrEnum
Binary health status for a component or aggregate report.
OK: the check passed. At the aggregate level: every critical check passed (non-critical failures do not flip the aggregate).ERROR: the check failed. At the aggregate level: at least one critical check failed.
OK
class-attribute
instance-attribute
OK = 'ok'
ERROR
class-attribute
instance-attribute
ERROR = 'error'
HealthSettingsValidationError
HealthSettingsValidationError(error: ValidationError | str)
Bases: HealthError, SettingsValidationError
Health Settings Validation Error.
Initialize from a Pydantic validation error.
details
instance-attribute
details = None