Python observability

Skill Mattakushi432/Claude-Code-Skills-Custom-DevTools-Pack/plugins/devtools-pack/skills/python-observability

A curated pack of custom Claude Code skills for developers — installable as a Claude Code plugin marketplace.

Install

npx -y skills add Mattakushi432/Claude-Code-Skills-Custom-DevTools-Pack --skill python-observability

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

0 stars0 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

What its author says it does

Copied from the file, not written here

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

SKILL.md

3.6 KB, 751 tokens by cl100k_base, as published. Nobody here has run it

Python Observability Patterns

OpenTelemetry Setup

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

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

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