Skip to content

Health

  • Start here: Health Checks guide
  • FastAPI integration: health_router for liveness, readiness, and health endpoints.
  • Every other framework: health_asgi mounts the same endpoints as a pure-ASGI app, and OpsServer serves them on a port of its own.

grelmicro.health

Health Checks.

Each name loads its submodule on first access.

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.

ProviderChecks 

ProviderChecks = Literal['off', 'report', 'require']

Which check HealthChecks registers for each Provider on the app.

"off" registers none. "report" registers a non-critical provider:{short_name} check, shown in /healthz and left out of /readyz. "require" registers a critical one, so a down backend fails /readyz.

CheckResult

Bases: TypedDict

Result of a single health check.

status instance-attribute 

status: HealthStatus

critical instance-attribute 

critical: bool

error instance-attribute 

error: str | None

details instance-attribute 

details: HealthDetails | 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: int | timedelta | None = None,
    liveness: Liveness | None = None,
    provider_checks: ProviderChecks = "report",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

HealthChecks(
    *,
    name: str = "default",
    timeout: PositiveFloat | None = None,
    cache_ttl: int | timedelta | None = None,
    liveness: Liveness | None = None,
    provider_checks: None,
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

HealthChecks(
    *,
    name: str = "default",
    timeout: PositiveFloat | None = None,
    cache_ttl: int | timedelta | None = None,
    liveness: Liveness | None = None,
    provider_checks: ProviderChecks | None = "report",
    env_prefix: str | None = None,
    env_load: bool | None = None,
)

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 HealthChecks instances may coexist on one Grelmicro under different names.

TYPE: str DEFAULT: 'default'

timeout

Default per-check timeout in seconds. Checks that exceed this duration are reported as error.

Default: 5.0. When unset and env reads are enabled (see env_load and GREL_ENV_LOAD), resolves from the environment variable GREL_HEALTH_TIMEOUT (or GREL_HEALTH_{NAME_UPPER}_TIMEOUT for a named instance) if present, otherwise falls back to the HealthChecksConfig default.

TYPE: PositiveFloat | None DEFAULT: None

cache_ttl

Per-check cache TTL, in whole seconds or as a timedelta. A float is refused. Set to 0 to disable.

Default: 1 second. When unset and env reads are enabled (see env_load and GREL_ENV_LOAD), resolves from the environment variable GREL_HEALTH_CACHE_TTL (or GREL_HEALTH_{NAME_UPPER}_CACHE_TTL for a named instance) if present, otherwise falls back to the HealthChecksConfig default.

TYPE: int | timedelta | None DEFAULT: None

liveness

When the worker counts as stuck, and exits so it is replaced. None (the default) runs no liveness check and starts no watchdog, so /livez always answers 200.

TYPE: Liveness | None DEFAULT: None

provider_checks

Which check to register for each Provider active on the app, on startup, matched in any case. "report" (the default) registers a non-critical provider:{short_name} check, shown in /healthz and left out of /readyz. "require" registers a critical one, so a down backend fails /readyz, and needs an active app on startup. "off" registers none. A Provider already added with add_provider keeps that registration.

None reads as "report" and warns with Grelmicro045Warning. It is removed in 0.45.0.

TYPE: ProviderChecks | None DEFAULT: 'report'

env_prefix

Override the auto-derived environment variable prefix.

Default: GREL_HEALTH_ for the default instance, GREL_HEALTH_{NAME_UPPER}_ for a named one.

TYPE: str | None DEFAULT: None

env_load

Whether to read environment variables.

When None (the default), follow the process-wide GREL_ENV_LOAD flag. Pass True or False to override the flag for this construction.

Pass False when the values here are the whole truth. Env reads fill every field not passed, so a config half taken from somewhere else silently gets the rest from the environment.

TYPE: bool | None DEFAULT: None

RAISES DESCRIPTION
SettingsValidationError

If provider_checks names no mode.

kind class-attribute 

kind: str = 'health'

name property 

name: str

Return the registration name.

is_alive property 

is_alive: bool

Whether the last round of liveness checks passed.

/livez answers 503 while this is false. It is also false once the liveness checks stopped on an unexpected error, which stops the worker. It is always true when no Liveness is set.

from_config classmethod 

from_config(
    config: HealthChecksConfig,
    *,
    name: str = "default",
    provider_checks: ProviderChecks = "report",
    liveness: Liveness | None = None,
) -> Self

from_config(
    config: HealthChecksConfig,
    *,
    name: str = "default",
    provider_checks: None,
    liveness: Liveness | None = None,
) -> Self

from_config(
    config: HealthChecksConfig,
    *,
    name: str = "default",
    provider_checks: ProviderChecks | None = "report",
    liveness: Liveness | None = None,
) -> 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 pydantic-settings aggregator). The environment path is bypassed and the config is used as-is.

TYPE: HealthChecksConfig

name

Registration name. Defaults to 'default'.

TYPE: str DEFAULT: 'default'

provider_checks

Which check to register for each active Provider on startup, as in HealthChecks. Defaults to "report". None reads as "report" and warns with Grelmicro045Warning. It is removed in 0.45.0.

TYPE: ProviderChecks | None DEFAULT: 'report'

liveness

When the worker counts as stuck, and exits so it is replaced. None (the default) starts nothing.

TYPE: Liveness | None DEFAULT: None

RAISES DESCRIPTION
SettingsValidationError

If provider_checks names no mode.

add 

add(
    name: str,
    func: HealthCheckFunc,
    *,
    critical: bool = True,
    timeout: PositiveFloat | None = None,
    liveness: bool = False,
) -> None

Register a health check function.

PARAMETER DESCRIPTION
name

Unique name identifying this check.

TYPE: str

func

Async function: returns None or a details dict on success, raises on failure.

TYPE: HealthCheckFunc

critical

Whether this check affects the aggregate status and HTTP response code. Critical failures flip the aggregate to error and cause /readyz / /healthz to return 503. Non-critical failures are visible in the /healthz body but do not flip the aggregate. For a liveness check, only a critical failure counts toward failure_threshold, turns /livez to 503 and stops the worker. A non-critical one is logged like any failed check, and nothing more.

TYPE: bool DEFAULT: True

timeout

Per-check timeout override. Falls back to the default when omitted.

TYPE: PositiveFloat | None DEFAULT: None

liveness

Run this check for /livez instead of /readyz and /healthz. A liveness check looks inside the process only, such as whether a consumer made progress, never at a database. It runs every Liveness.interval seconds. Refused when HealthChecks(liveness=...) is not set.

TYPE: bool DEFAULT: False

RAISES DESCRIPTION
ValueError

If name is already registered, or does not match ^[a-z0-9][a-z0-9:_-]*$ (max 64 chars), or liveness=True is set on a HealthChecks with no Liveness, or timeout is not a finite number of seconds greater than zero. A colon is allowed in name for namespacing, e.g. "weather:circuitbreaker".

check 

check(
    name: str,
    *,
    critical: bool = True,
    timeout: PositiveFloat | None = None,
    liveness: bool = False,
) -> Callable[[FuncT], FuncT]

Decorate an async function to register it as a health check.

Registration is a side effect: the function is returned unchanged with its original signature, so await-ing it directly (for example in a unit test) type-checks as usual.

PARAMETER DESCRIPTION
name

Unique name identifying this check.

TYPE: str

critical

Whether this check affects the aggregate status. For a liveness check, whether its failures can stop the worker.

TYPE: bool DEFAULT: True

timeout

Per-check timeout override.

TYPE: PositiveFloat | None DEFAULT: None

liveness

Run this check for /livez instead of /readyz and /healthz.

TYPE: bool DEFAULT: False

Example

@health.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: Provider

name

Check name suffix. Defaults to the provider's short_name, so the check is provider:redis. Pass an explicit name to disambiguate two providers of the same vendor, e.g. name='sessions' registers provider:sessions.

TYPE: str | None DEFAULT: None

critical

Whether the check affects /readyz. Critical by default: an unreachable backend fails readiness. Pass critical=False for a degradable dependency such as a cache.

TYPE: bool DEFAULT: True

timeout

Per-check timeout override. Falls back to the default.

TYPE: PositiveFloat | None DEFAULT: None

RAISES DESCRIPTION
ValueError

If the provider ships no readiness check, the resulting name is already registered, or timeout is not a number of seconds greater than zero.

add_circuitbreaker 

add_circuitbreaker(
    breaker: CircuitBreaker,
    *,
    name: str | None = None,
    critical: bool = False,
    timeout: PositiveFloat | None = None,
) -> None

Register a check on a circuit breaker as circuitbreaker:{name}.

Each run reads the breaker's state from its backend, so a replica that made no call through a shared breaker still reports it open. A closed breaker is ok. A half-open or forced-closed breaker is ok with its state in the details. An open breaker fails with the seconds until its next probe, and a forced-open one fails until it is reset.

PARAMETER DESCRIPTION
breaker

The circuit breaker to report. A per-key circuit from breaker.keyed(key) is reported on its own.

TYPE: CircuitBreaker

name

Check name suffix. Defaults to the breaker's name, so the check is circuitbreaker:inverter. Pass an explicit name when the breaker name is not a valid check name.

TYPE: str | None DEFAULT: None

critical

Whether an open breaker fails /readyz. Non-critical by default: an open breaker shows in the /healthz body and leaves readiness alone.

TYPE: bool DEFAULT: False

timeout

Per-check timeout override. Falls back to the default.

TYPE: PositiveFloat | None DEFAULT: None

RAISES DESCRIPTION
ValueError

If the resulting name is not a valid check name (pass name=), is already registered, or timeout is not a number of seconds greater than zero.

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. Concurrent calls for the same check coalesce via single-flight.

PARAMETER DESCRIPTION
critical_only

If True, only run critical checks.

TYPE: bool DEFAULT: False

exclude

Check names to skip.

TYPE: Iterable[str] | None DEFAULT: None

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: float = 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: Retention = timedelta(seconds=1)

Per-check cache TTL, in whole seconds or as a timedelta. A float is refused. Each check's last result is reused until it is cache_ttl old. Concurrent calls coalesce via single-flight. Set to 0 to disable caching. From text, such as an environment variable, it reads whole seconds ("60") or an ISO 8601 duration ("PT0.5S").

HealthReport

Bases: TypedDict

Aggregated health report across all registered checks.

status instance-attribute 

status: HealthStatus

checks instance-attribute 

checks: dict[str, CheckResult]

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'

Liveness

Bases: BaseModel

When a worker counts as stuck, and exits so it is replaced.

A stuck worker exits, and the process manager that runs it starts another: the uvicorn supervisor, the Gunicorn arbiter, or Kubernetes for a container whose only process ends. A loop that stalls past stall_timeout exits at once. Liveness checks that fail failure_threshold rounds in a row stop the worker with SIGTERM, and /livez answers 503 from the first failed round.

stall_timeout class-attribute instance-attribute 

stall_timeout: float | None = None

Seconds the event loop may go without running a callback. A watchdog thread checks four times per stall_timeout, at most once a second. Past it, the watchdog logs the loop thread's stack and exits the process. None (the default) starts no watchdog.

interval class-attribute instance-attribute 

interval: float = 10.0

Seconds between two runs of the liveness checks.

failure_threshold class-attribute instance-attribute 

failure_threshold: PositiveInt = 3

Liveness rounds that must fail in a row before the worker stops itself with SIGTERM. /livez answers 503 from the first one.

shutdown_timeout class-attribute instance-attribute 

shutdown_timeout: float = 30.0

Seconds a worker stopped by failed liveness checks has to shut down. Past it, the worker exits at once, as it does when it runs as PID 1 with no SIGTERM handler or its shutdown hangs.

health_asgi 

health_asgi(
    component: HealthChecks | None = None,
    *,
    prefix: str = "",
    show_details: bool = False,
) -> ASGIApp

Create a pure-ASGI app serving the health endpoints.

The three endpoints health_router serves, rendered by the same code, with no framework anywhere:

  • GET/HEAD {prefix}/livez: Liveness probe. Runs no check per request. 503 while a liveness check fails, else 200, with an empty body.
  • GET/HEAD {prefix}/readyz: Readiness probe. Runs critical checks only. 200 or 503 with an empty body.
  • GET/HEAD {prefix}/healthz: Aggregate JSON report.

Both probe paths accept ?exclude=name,name, and every response sets Cache-Control: no-store. Any other path answers 404, and any other method 405.

Mount it in an ASGI framework:

from starlette.applications import Starlette
from starlette.routing import Mount

from grelmicro.health import health_asgi

app = Starlette(routes=[Mount("", app=health_asgi())])

Or give it a port of its own with OpsServer, for a worker that runs no web framework at all.

On FastAPI, prefer health_router(): it serves the same endpoints and adds the OpenAPI schema, the Depends gate on /healthz, and the dependency form of show_details.

PARAMETER DESCRIPTION
component

Health checks component whose checks the app runs. When omitted, it resolves the default instance from the active Grelmicro app (Grelmicro(uses=[HealthChecks(...)])).

TYPE: HealthChecks | None DEFAULT: None

prefix

URL prefix for the three paths. Leave it empty when the app is mounted, because the mount strips its own path first.

TYPE: str DEFAULT: ''

show_details

Whether /healthz includes each check's verbose details field (versions, hostnames, pool stats, ...). False (the default) strips them, which is safe for an endpoint anyone can reach.

TYPE: bool DEFAULT: False