Health checks¶
alchemiq provides async health probes for all configured backends (PostgreSQL,
ClickHouse, and the cache layer). The probes run concurrently and collect
latency measurements, making the result suitable for Kubernetes
readinessProbe and livenessProbe endpoints.
The public API is importable directly from alchemiq.health:
from alchemiq.health import check_health, HealthReport, ComponentHealth
Running a health check¶
check_health() is an async function. Call it from any async
context - a FastAPI lifespan, a background task, or a manual probe:
from alchemiq.health import check_health
report = await check_health()
if not report.healthy:
print(report.to_dict())
The function probes every configured backend concurrently. Backends that are
not configured are silently skipped, so the same call works whether you have
one database or three. When no backends are configured at all (e.g. in unit
tests that never call configure), check_health returns a trivially
healthy report with an empty components tuple. An opt-in strict-readiness
mode that treats “no backend configured” as unhealthy is on the roadmap (see
What’s not in v1).
Timeout¶
Each probe is guarded by a per-call timeout (default five seconds). Pass a different value when a tighter budget is appropriate:
report = await check_health(timeout=2.0)
A probe that times out or raises an exception is recorded as unhealthy rather than propagating the error to the caller.
Return shape¶
HealthReport¶
The aggregate result returned by check_health().
Field |
Type |
Description |
|---|---|---|
|
|
|
|
|
One entry per probed backend |
report.to_dict()
# {
# "status": "healthy", # or "unhealthy"
# "checks": [
# {"name": "postgres", "healthy": True, "latency_ms": 1.4, "error": None},
# {"name": "clickhouse", "healthy": False, "latency_ms": None, "error": "..."},
# {"name": "cache", "healthy": True, "latency_ms": 0.3, "error": None},
# ]
# }
ComponentHealth¶
The probe result for a single backend component.
Field |
Type |
Description |
|---|---|---|
|
|
Component identifier: |
|
|
Whether the probe succeeded |
|
|
Round-trip latency in milliseconds; |
|
|
Short error description when unhealthy; |
Both classes are frozen dataclasses (immutable, __slots__).
FastAPI integration¶
The [fastapi] extra provides health_router, a pre-built APIRouter
that mounts two endpoints:
Endpoint |
Kubernetes probe |
Behaviour |
|---|---|---|
|
|
Calls |
|
|
Always returns 200 |
Include the router once at application startup:
from fastapi import FastAPI
from alchemiq.fastapi import health_router
app = FastAPI()
app.include_router(health_router())
The router accepts optional keyword arguments:
Parameter |
Default |
Description |
|---|---|---|
|
|
URL prefix for both routes |
|
|
Per-probe timeout passed to |
|
|
Set to |
# Custom prefix:
app.include_router(health_router(prefix="/probe"))
# Readiness only (no liveness route):
app.include_router(health_router(include_liveness=False))
Liveness vs readiness¶
The two probe types serve different purposes in Kubernetes:
Readiness (/health/ready) answers “can this pod serve traffic right now?”
A pod failing readiness is removed from the load-balancer pool but is not
restarted. Use it to signal that a required database or cache is unreachable.
Liveness (/health/live) answers “is this pod still alive?” A pod failing
liveness is restarted by the kubelet. The liveness route in alchemiq is
intentionally dependency-free - it just confirms the event loop is running -
because a liveness failure that triggers a restart would not fix a broken
database connection.
A typical Kubernetes deployment manifest:
readinessProbe:
httpGet:
path: /health/ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /health/live
port: 8000
initialDelaySeconds: 10
periodSeconds: 30