Step 7 — Observability

A server is only valuable if it's alive. Knowing whether it's alive is observability.

Health endpoints

@router.get("/health")
def health(): return {"status": "ok"}

@router.get("/health/db")
def health_db():
    try:
        with get_conn() as conn, conn.cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()
        return {"db": "ok"}
    except Exception as e:
        return {"db": "down", "error": str(e)}, 503

/health = process up; /health/db = up to and including the DB.

Docker HEALTHCHECK

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

Structured logging

import logging, json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "ts": self.formatTime(record),
            "level": record.levelname,
            "msg": record.getMessage(),
            "module": record.module,
        })

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.getLogger().addHandler(handler)

JSON logs are auto-parseable by Loki / CloudWatch / Datadog.

Metrics — Prometheus

from prometheus_client import Counter, Histogram, generate_latest

requests_total = Counter("http_requests_total", "total requests", ["method", "path"])

@router.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type="text/plain")

Grafana scrapes /metrics once per minute.

Alerts — three things to decide

• What — DB connection failed / 5xx error rate >1% / memory >80%
• Who — Slack channel, email
• When — weekday 9–18 only, or 24/7

Try it

Add /health + /health/db and a Dockerfile HEALTHCHECK. docker compose ps should show STATUS healthy.

Next

The final step 8 gathers the routers, async, validation, dependencies, and error handling so far into FastAPI patterns that do not break in practice.