Observability Basics
10 examples to get you started with Observability - 7 basic and 3 intermediate.
//
Search across all documentation pages
10 examples to get you started with Observability - 7 basic and 3 intermediate.
uv venv && source .venv/bin/activate
uv pip install "structlog>=24.0" "prometheus-client>=0.21" "opentelemetry-sdk>=1.29"JSON logs are searchable in Loki, CloudWatch, and ELK.
import json
import logging
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("api")
log.info(json.dumps({"event": "order_created", "order_id": "ord_1", "ms": 42}))order_id) not only message stringsRelated: Structured Logging - structlog setup
Tie log lines to a single HTTP request.
import uuid
from contextvars import ContextVar
request_id: ContextVar[str] = ContextVar("request_id", default="-")
def set_request_id() -> str:
rid = str(uuid.uuid4())
request_id.set(rid)
return ridX-Request-ID from gateway or generate at edgerequest_id on every log record in that requestRelated: Distributed Tracing - trace IDs
Count events for rate and error ratio dashboards.
from prometheus_client import Counter
ORDERS_CREATED = Counter("orders_created_total", "Orders created")
def create_order() -> None:
ORDERS_CREATED.inc()_total is Prometheus convention/metrics endpoint for scrapingRelated: Metrics - histograms and gauges
Measure how long operations take.
import time
start = time.perf_counter()
# ... work ...
elapsed_ms = (time.perf_counter() - start) * 1000perf_counter monotonic for durationsRelated: Performance Monitoring (APM) - latency SLOs
Load balancers need a cheap OK response.
from fastapi import FastAPI
app = FastAPI()
@app.get("/health/live")
def live():
return {"status": "ok"}Related: Health & Readiness - probe design
Use levels to separate signal from noise.
import logging
log = logging.getLogger("worker")
log.debug("detail for dev")
log.info("job finished")
log.warning("retrying")
log.error("failed permanently")Related: Observability Best Practices - logging rules
Capture stack traces with context.
import logging
log = logging.getLogger("api")
try:
risky()
except Exception:
log.exception("payment failed", extra={"order_id": "1"})log.exception includes traceback automaticallyextra dict (JSON formatter must include it)Related: Error Tracking - Sentry
Rate, Errors, Duration per endpoint.
# Rate: requests_total counter
# Errors: requests_failed_total counter
# Duration: request_duration_seconds histogramRelated: Metrics - histogram buckets
One trace ID, multiple spans across services.
[API span] -> [DB span] -> [SQS publish span]
Related: Distributed Tracing - OpenTelemetry
Incident flow: metric alert → logs with request_id → trace waterfall.
# 1. Alert: error_rate > 1%
# 2. Log query: request_id="abc" AND level=ERROR
# 3. Trace UI: trace_id from log fieldtrace_id, request_id) across pillarsRelated: Observability Best Practices - on-call hygiene
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.
Reviewed by Chris St. John·Last updated Jul 16, 2026