Skip to content

Health

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.

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: 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 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 seconds. Set to 0 to disable.

Default: 1.0. 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: NonNegativeFloat | None DEFAULT: None

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.

TYPE: bool | None DEFAULT: None

auto_health

Register a provider:{short_name} readiness check for every Provider active on the app, on startup. Off by default. Each registered check is critical, so an unreachable backend fails /readyz. For finer control, leave this off and call add_provider per provider.

TYPE: bool DEFAULT: False

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 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'

auto_health

Register a provider:{short_name} readiness check for every active Provider on startup. Off by default.

TYPE: bool DEFAULT: False

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: 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.

TYPE: bool DEFAULT: True

timeout

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

TYPE: PositiveFloat | None DEFAULT: None

RAISES DESCRIPTION
ValueError

If name is already registered, or does not match ^[a-z0-9][a-z0-9:_-]*$ (max 64 chars). Colon is allowed for namespacing, e.g. "weather:circuitbreaker".

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: str

critical

Whether this check affects the aggregate status.

TYPE: bool DEFAULT: True

timeout

Per-check timeout override.

TYPE: PositiveFloat | None DEFAULT: None

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: 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, 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: 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: 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.

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'

HealthSettingsValidationError

HealthSettingsValidationError(error: ValidationError | str)

Bases: HealthError, SettingsValidationError

Health Settings Validation Error.

Initialize from a Pydantic validation error.

details instance-attribute

details = None