# Python Observability

> When to activate: OpenTelemetry, traces, metrics, spans, Prometheus, Grafana, Sentry, APM setup

- Skill: `mattakushi432/python-observability` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/python-observability`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/python-observability/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/python-observability

---


# Python Observability Patterns

## OpenTelemetry Setup
```python
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

def configure_tracing(app_name: str, otlp_endpoint: str) -> None:
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    
    # Auto-instrument popular libraries
    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument()
    HTTPXClientInstrumentor().instrument()

# Manual spans
tracer = trace.get_tracer(__name__)

async def process_order(order_id: str) -> dict:
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order_id)
        
        with tracer.start_as_current_span("validate_inventory"):
            items = await inventory_service.check(order_id)
            span.set_attribute("items.count", len(items))
        
        result = await payment_service.charge(order_id)
        span.set_attribute("payment.status", result["status"])
        return result
```

## Prometheus Metrics
```python
from prometheus_client import Counter, Histogram, Gauge, make_asgi_app
import time

# Define metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status_code"],
)

REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)

ACTIVE_CONNECTIONS = Gauge("active_connections", "Active WebSocket connections")

# Middleware
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.monotonic()
    response = await call_next(request)
    duration = time.monotonic() - start
    
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status_code=response.status_code,
    ).inc()
    
    REQUEST_DURATION.labels(
        method=request.method,
        endpoint=request.url.path,
    ).observe(duration)
    
    return response

# Expose metrics endpoint
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)
```

## Sentry Error Tracking
```python
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

sentry_sdk.init(
    dsn=settings.sentry_dsn,
    environment=settings.environment,
    traces_sample_rate=0.1,  # 10% of transactions
    profiles_sample_rate=0.1,
    integrations=[FastApiIntegration(), SqlalchemyIntegration()],
    before_send=scrub_pii,  # remove PII before sending
)

def scrub_pii(event: dict, hint: dict) -> dict | None:
    if "request" in event:
        # Remove auth headers
        event["request"].get("headers", {}).pop("Authorization", None)
    return event
```