Liveness, readiness and dependency health checks for FastAPI in three lines. Probes run concurrently, each with its own timeout, and results are cached so an orchestrator polling every second does not become a load test on your database.
from fastapi import FastAPI
from fastapi_readyz import HealthCheck, SqlAlchemyProbe, RedisProbe, HttpProbe, add_health_routes
app = FastAPI()
health = HealthCheck(
[
SqlAlchemyProbe("postgres", engine),
RedisProbe("redis", redis_client),
HttpProbe("payments-api", "https://payments.internal/ping"),
],
timeout=2.0, # per probe
cache_ttl=5.0, # reuse the report for 5 s
non_critical=["redis"], # redis down => degraded, not unhealthy
)
add_health_routes(app, health)You now have:
| Endpoint | Purpose | Status codes |
|---|---|---|
GET /health/live |
Process is up. Never touches dependencies. | always 200 |
GET /health/ready |
Safe to route traffic here? | 200 ok, 503 unhealthy |
GET /health |
Full report with per-probe latency and errors. | same as /ready |
{
"status": "degraded",
"duration_ms": 12.4,
"checked_at": 1788000000.1,
"checks": {
"postgres": {"status": "healthy", "latency_ms": 3.1, "details": {"dialect": "postgresql"}},
"redis": {"status": "unhealthy", "latency_ms": 2000.3, "error": "timeout after 2.0s"},
"payments-api": {"status": "healthy", "latency_ms": 9.8, "details": {"status_code": 200}}
}
}Most examples on the internet put SELECT 1 behind /health and call it a day.
That breaks in production in three predictable ways:
- Liveness and readiness get conflated. A slow database makes Kubernetes
restart a perfectly healthy pod. Here
/livenever touches dependencies. - One slow probe blocks the rest. Probes here run concurrently under
asyncio.gather, each wrapped in its ownwait_fortimeout. - Every poll hits every dependency. With
cache_ttla burst of concurrent readiness calls triggers exactly one round of probes; the others wait for the same result.
And a fourth, less obvious one: not every dependency is critical. A dead cache
should show up as degraded in your dashboards without pulling the service out
of the load balancer. non_critical handles that.
pip install fastapi-readyz # core, no extra deps beyond FastAPI
pip install "fastapi-readyz[http]" # + httpx for HttpProbe
pip install "fastapi-readyz[sqlalchemy,redis]"| Probe | Checks |
|---|---|
SqlAlchemyProbe |
SELECT 1 on a sync Engine or AsyncEngine |
RedisProbe |
PING on redis-py, sync or async client |
HttpProbe |
GET a URL, require status in expect |
TcpProbe |
TCP connect to host:port |
DiskSpaceProbe |
free space on a path above min_free_mb |
CallableProbe |
any function or coroutine; raise to fail |
No base class. Anything with a name and an async check() works:
class QueueDepthProbe:
name = "queue"
def __init__(self, queue, max_depth: int = 10_000):
self.queue, self.max_depth = queue, max_depth
async def check(self):
depth = await self.queue.size()
if depth > self.max_depth:
raise RuntimeError(f"queue backed up: {depth} messages")
return {"depth": depth} # shows up under "details"livenessProbe:
httpGet: { path: /health/live, port: 8000 }
periodSeconds: 10
readinessProbe:
httpGet: { path: /health/ready, port: 8000 }
periodSeconds: 5
failureThreshold: 3Set HealthCheck(cache_ttl=...) to something a bit below periodSeconds so the
report is fresh for every poll but never computed twice per period.
HealthCheck(probes, *, timeout=5.0, cache_ttl=0.0, non_critical=())
make_health_router(health, *, prefix="/health", tags=["health"], degraded_is_error=False)
degraded_is_error=True makes /ready return 503 for degraded too, if your
load balancer should stop sending traffic whenever anything is off.
pip install -e ".[dev]"
ruff check . && pytestTests run against real components where cheap: an in-process TCP listener,
SQLite through both sync and async SQLAlchemy engines, and an ASGI-mounted
upstream for HttpProbe. No network access needed.
MIT