Step 7
Step 7 — Observability
25 min
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
Course done. Try devops-cloud or ai-agent-tooling.
🎉 You finished Python · FastAPI · Data Pipelines
What's next? Pick another course below.