Health Checks

A health checks manager with concurrent execution and FastAPI integration. Use it to expose liveness, readiness, and aggregate health endpoints.

HealthChecks: Register check functions with a FastAPI-style decorator, run them concurrently with per-check timeouts and caching.
health_router: FastAPI router with /livez, /readyz, and /healthz endpoints.

Quick start

Create a HealthChecks and register checks with the @health.check(name) decorator:

from grelmicro.health import HealthChecks, HealthDetails

# Create the HealthChecks instance and register it with the Grelmicro
# app via `Grelmicro(uses=[health])` so the router can resolve it.
health = HealthChecks()


# Register checks with the @health.check(name) decorator
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success, raise on failure
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details
    return {"latency_ms": 1.2}


# Optional dependency: mark non-critical so its failure doesn't
# take the instance out of the load balancer.
@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    return None

Register the instance with a Grelmicro app (see below) so the router can resolve it, or pass it explicitly via health_router(registry=health).

Health Check

A health check is a function returning None (healthy) or a HealthDetails dict (also healthy, with metrics attached). Raising an exception signals failure.

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


# Decorator form: register an async function under a name
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success (healthy, no details)
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details (e.g. metrics)
    return {"latency_ms": 1.2, "version": "7.2"}


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    # Raise HealthError to expose a specific message in the error field.
    # Other exceptions produce an error formatted as "{ExceptionType}: {message}".
    msg = "Connection refused"
    raise HealthError(msg)

Return None: healthy, no details.
Return a HealthDetails dict: healthy, with details. Values can be primitives, datetime, nested dicts, lists, or tuples.
Raise HealthError: unhealthy. The exception message appears in the error field.
Raise any other exception: unhealthy, with an error field formatted as "{ExceptionType}: {message}" (for example "ValueError: bad input"). The traceback is logged server-side to avoid leaking internal information.

HealthDetails is a type alias for dict[str, JSONEncodable]. Both sync and async check functions are supported. Sync functions run in a worker thread via asyncio.to_thread so they never block the event loop.

Grelmicro app integration

Register the HealthChecks with a Grelmicro app to lifecycle it alongside the rest of your modules. Same FastAPI-style explicit registration as Tasks:

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks

health = HealthChecks()
micro = Grelmicro(uses=[health])

@health.check("redis")
async def redis_alive() -> None:
    ...

async with micro:
    report = await health.run()

Use Grelmicro.use(item) (or uses=) for entry-point components like HealthChecks and Tasks. The caller keeps the reference and uses it directly.

For imperative registration (without a decorator), use health.add(name, func):

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


async def check_kafka() -> HealthDetails | None:
    return None


health.add("kafka", check_kafka, critical=True, timeout=2.0)

Critical vs Non-Critical

By default, all checks are critical: a failure flips the aggregate to error and returns 503 on /readyz and /healthz. Pass critical=False for optional dependencies:

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    return None

Scenario	Aggregate status on `/healthz`	`/readyz`	`/healthz`
All critical pass	`ok`	`200`	`200`
Non-critical failed	`ok`	`200`	`200`
Critical failed	`error`	`503`	`503`

Non-critical checks do not pull the instance from the load balancer. They are not run on /readyz (which runs critical only). Their status appears per-check in the /healthz body so operators and dashboards can see degraded dependencies without triggering traffic removal.

Timeout

Checks that exceed their timeout are reported as failed:

from grelmicro.health import HealthChecks, HealthDetails

# Registry default: 2s per check (default is 5s)
health = HealthChecks(timeout=2.0)


# Per-check override: tight timeout for a flaky optional dep
@health.check("analytics", critical=False, timeout=0.5)
async def check_analytics() -> HealthDetails | None:
    return None

HealthChecks has a global default timeout (5.0 seconds). Per-check overrides are set on registration:

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


@health.check("slow-api", critical=False, timeout=0.5)
async def check_slow_api() -> HealthDetails | None:
    return None

A slow non-critical check hits the timeout and is reported with status: "error" in the response body, but the aggregate stays ok and /readyz stays 200.

Timeout detection uses asyncio.timeout. The wrapper distinguishes the configured timeout from a TimeoutError raised inside the check itself (for example a socket timeout).

Caching

HealthChecks caches each check's result for cache_ttl seconds (default 1.0) and coalesces concurrent calls via single-flight per check. A given check runs at most once per TTL regardless of how many endpoints or concurrent requests are in flight. This prevents probe traffic from amplifying onto your database.

from grelmicro.health import HealthChecks

# Default: 1-second TTL with single-flight per check
HealthChecks(timeout=5.0, cache_ttl=1.0)

# Disable caching entirely
HealthChecks(cache_ttl=0)

Returning Details with an Error

Raise HealthError(message, details=...) to attach a diagnostic payload to a failing check. The payload appears under details on the check entry, subject to show_details:

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


@health.check("database")
async def check_database() -> HealthDetails | None:
    # Simulate a failure with a diagnostic payload
    msg = "connection pool exhausted"
    raise HealthError(msg, details={"active": 10, "idle": 0, "max": 10})

Provider Readiness Checks

Every connection provider ships a built-in readiness probe. Redis and Valkey run PING, Postgres and SQLite run SELECT 1. These are cheaper and more honest than a borrowed lock round trip, and you no longer hand-write one check per backend.

Register a provider's check with add_provider. It registers under provider:{short_name}, for example provider:redis:

from grelmicro.health import HealthChecks
from grelmicro.providers.redis import RedisProvider

store = RedisProvider("redis://localhost:6379")
cache = RedisProvider("redis://localhost:6379/1")

health = HealthChecks()

# Register the provider's built-in readiness check as "provider:redis".
health.add_provider(store)

# A degradable dependency: visible in /healthz, never fails /readyz.
health.add_provider(cache, name="cache", critical=False)

The check is critical by default, so an unreachable backend fails /readyz. Pass critical=False for a degradable dependency such as a cache. Pass name= to disambiguate two providers of the same vendor, which registers provider:{name} instead.

To register a check for every provider on the app at once, build the HealthChecks with auto_health=True:

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.providers.postgres import PostgresProvider
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379")
postgres = PostgresProvider("postgresql://localhost:5432/app")

# Registers "provider:redis" and "provider:postgres" critical checks on startup.
micro = Grelmicro(uses=[redis, postgres, HealthChecks(auto_health=True)])

On startup this discovers every active provider, including ones a Component borrows without listing, and registers a critical provider:{short_name} check for each. A provider with no built-in probe is skipped. It is off by default, so checks appear only when you ask for them. When two providers share a short_name, the second is skipped with a warning: register it explicitly with add_provider(provider, name=...) to give it a distinct name.

FastAPI Integration

Add health endpoints to your FastAPI app:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.integrations.fastapi import health_router

health = HealthChecks()


@health.check("database")
async def check_database() -> HealthDetails | None:
    return None


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    yield


app = FastAPI(lifespan=lifespan)
app.include_router(health_router())
# Endpoints: GET /livez, GET /readyz, GET /healthz

This creates three endpoints:

Endpoint	Purpose	Success	Failure	Body
`GET /livez`	Liveness probe. Never runs checks.	`200`	no response (timeout)	empty
`GET /readyz`	Readiness probe. Runs critical checks only.	`200`	`503`	empty
`GET /healthz`	Aggregate JSON report for humans and dashboards. Runs all checks.	`200`	`503`	JSON `{status, checks}`

All three also accept HEAD. All responses set Cache-Control: no-store. Probe endpoints return an empty body. The HTTP status code is the entire signal.

Paths follow the z-pages convention (/livez, /readyz, /healthz). The trailing z avoids collisions with application routes like /health.

Using with Docker, Compose, and other Orchestrators

Different orchestrators consume different probes. grelmicro exposes all three endpoints, pick the ones that fit:

Orchestrator	Uses
Kubernetes, OpenShift	`livenessProbe` → `/livez`, `readinessProbe` → `/readyz`, `startupProbe` → `/livez`
Docker (`HEALTHCHECK`), Docker Compose, Docker Swarm	single healthcheck → `/livez` (restart on failure)
Nomad (`check` stanza)	`/livez` for liveness, `/readyz` for routing
systemd (`WatchdogSec`)	`/livez` (restart on failure)
Reverse proxies (Traefik, nginx, HAProxy, Envoy), load balancers (AWS ALB, GCP LB)	`/readyz` for upstream health
Dashboards, uptime monitors (Prometheus, Pingdom)	`/healthz` for full report

Docker Compose example:

services:
  app:
    image: myapp
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/livez"]
      interval: 10s
      timeout: 2s
      retries: 3

Docker and Docker Swarm only model a single healthcheck per container, so point it at /livez. Route readiness through a reverse proxy that probes /readyz.

Exclude Checks

Both /readyz and /healthz accept an ?exclude query parameter with a comma-separated list of check names. Useful for temporarily muting a known-flaky check without redeploy:

GET /readyz?exclude=analytics,recommendations
GET /healthz?exclude=analytics

Excluded checks are not run and do not appear in the response.

Verbose Details

Each check can return verbose metadata (versions, internal hostnames, pool stats, latencies). Exposing it publicly on /healthz leaks infrastructure fingerprints to anyone who can reach the endpoint, so by default grelmicro strips every details field from the response.

The show_details parameter controls visibility with three forms:

`show_details`	Who sees details	Use when
`False` (default)	nobody	Safe default for a public `/healthz`
`True`	everyone	`/healthz` is on a private network only
`Depends(fn)`	requests for which `fn()` returns `True`	Public `/healthz`, admin-only details

With show_details=Depends(fn), fn is wired into FastAPI's dependency-injection graph. Return True to include details for that request, False to strip them. Everything FastAPI supports works: Depends sub-dependencies, Security, Request injection, async functions, yield cleanup. Returning False strips details but the endpoint still returns 200/503 with status and check names. This way, uptime monitors without credentials still get actionable aggregate status while admin tools with credentials get the full payload. Raising HTTPException blocks the endpoint, so prefer returning False for a soft strip.

from ipaddress import ip_address

from fastapi import Depends, Request

from grelmicro.integrations.fastapi import health_router


def from_private_network(request: Request) -> bool:
    return bool(request.client and ip_address(request.client.host).is_private)


# Hide details from everyone (default).
router = health_router()

# Show details to everyone (private /healthz only).
router = health_router(show_details=True)

# Show details when the dependency returns True.
router = health_router(show_details=Depends(from_private_network))

With details enabled, each check entry includes a details field:

{
  "status": "ok",
  "checks": {
    "redis": {
      "status": "ok",
      "critical": true,
      "error": null,
      "details": {"latency_ms": 1.2, "version": "7.2"}
    }
  }
}

The error field is always returned

show_details strips only the details field. Each failing check still reports its error string, even on a public /healthz. Keep check error messages safe for the endpoint audience: raise HealthError with a short, non-sensitive message and put diagnostics in details, which stays hidden by default. To hide error strings too, gate the whole endpoint with healthz_dependencies.

`show_details` vs `healthz_dependencies`

Two independent gates sit in front of /healthz:

Parameter	Failure effect	Typical use
`healthz_dependencies`	Blocks the entire endpoint (`401`/`403`)	Keep `/healthz` entirely private
`show_details=Depends(fn)`	When `fn()` returns `False`, strips the `details` field. Endpoint still returns `200`/`503`	Keep aggregate status public, hide verbose metadata

For stricter setups, gate the whole /healthz endpoint behind authentication while leaving /livez and /readyz open (most orchestrators and load balancers cannot carry credentials):

from fastapi import Depends, Request

from grelmicro.integrations.fastapi import health_router


def require_admin(request: Request) -> None:
    # Replace with your real auth. Raise HTTPException on failure.
    return None


# Gate /healthz behind auth. /livez and /readyz remain open for
# orchestrators and load balancers.
router = health_router(
    show_details=True,
    healthz_dependencies=[Depends(require_admin)],
)

URL Prefix

Mount the health endpoints under a custom prefix:

from fastapi import FastAPI

from grelmicro.integrations.fastapi import health_router

app = FastAPI()
app.include_router(health_router(prefix="/api/v1"))
# Endpoints: GET /api/v1/livez, GET /api/v1/readyz, GET /api/v1/healthz

Design

Why Three Endpoints

Each endpoint serves a different audience:

Endpoint	Audience	Answers
`/livez`	Orchestrator (Kubernetes, Docker, Nomad, systemd)	Is the process alive? Should it be restarted?
`/readyz`	Load balancer, reverse proxy, service mesh	Can this instance serve traffic?
`/healthz`	Operators, dashboards, uptime monitors	What is the state of each component?

Liveness never checks dependencies. A failing database must never restart your container. Readiness runs all critical checks concurrently. If any fails, the instance is removed from the load balancer. Aggregate also runs non-critical checks and returns a JSON report for humans. Probe bodies stay empty to keep the wire minimal and the signal unambiguous. The HTTP status code is the only thing orchestrators read.

Status Vocabulary

Binary status, used for both components and the aggregate:

Status	Meaning
`ok`	The check passed. At the aggregate level: every critical check passed.
`error`	The check failed. At the aggregate level: at least one critical check failed.

Non-critical failures produce status: "error" on the individual check but do not flip the aggregate. The aggregate only goes to error when at least one critical check fails.

Function-Based API

Checks are plain functions. No base class to inherit from.

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


# Decorator form: register an async function under a name
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success (healthy, no details)
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details (e.g. metrics)
    return {"latency_ms": 1.2, "version": "7.2"}


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    # Raise HealthError to expose a specific message in the error field.
    # Other exceptions produce an error formatted as "{ExceptionType}: {message}".
    msg = "Connection refused"
    raise HealthError(msg)

This mirrors FastAPI's own @app.get("/path") routing style and keeps grelmicro small. Return types are statically checked. JSONEncodable (recursive) includes primitives, datetime, nested dict / list / tuple, and any Mapping subclass. Type checkers (mypy, ty) catch non-serializable returns like bytes or custom objects at authoring time.

Concurrent Execution

All selected checks run in parallel via an asyncio.TaskGroup. A slow check does not block other checks. Each check runs with its own timeout (falling back to the default).