A typed health check for FastAPI services

Sample recipe to show the format. Each recipe is a Markdown file in src/content/recipes/.

Kubernetes wants two answers: is the process alive and can it take traffic. Keep them separate so a slow dependency doesn’t get your pod killed.

from fastapi import FastAPI, Response, status
from pydantic import BaseModel


class Health(BaseModel):
    """Probe response body."""

    status: str
    checks: dict[str, bool] = {}


app = FastAPI()


async def db_ok() -> bool:
    """Return True if the database answers a trivial query."""
    return True  # replace with `SELECT 1`


@app.get("/livez", response_model=Health)
async def livez() -> Health:
    """Liveness: the event loop is responsive."""
    return Health(status="ok")


@app.get("/readyz", response_model=Health)
async def readyz(response: Response) -> Health:
    """Readiness: downstream dependencies are reachable."""
    checks = {"db": await db_ok()}
    if not all(checks.values()):
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return Health(status="degraded", checks=checks)
    return Health(status="ok", checks=checks)

And the test:

from fastapi.testclient import TestClient

from app import app


def test_readyz_ok() -> None:
    resp = TestClient(app).get("/readyz")
    assert resp.status_code == 200
    assert resp.json()["status"] == "ok"