Python Monitoring Guide: Django, FastAPI & Application Observability (2026)

# Install
pip install prometheus-fastapi-instrumentator prometheus-client

# main.py
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import Counter, Histogram, Gauge
import time

app = FastAPI()

# Auto-instrument all routes (adds http_request_duration_seconds,
# http_requests_total, http_request_size_bytes)
Instrumentator().instrument(app).expose(app)

# Custom business metrics
orders_total = Counter(
    "orders_total",
    "Total orders placed",
    ["payment_method", "status"]
)

payment_processing_seconds = Histogram(
    "payment_processing_seconds",
    "Time spent processing payments",
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

active_users = Gauge(
    "active_users",
    "Currently active users in the system"
)

@app.post("/orders")
async def create_order(order: OrderRequest):
    start = time.time()
    try:
        result = await process_order(order)
        orders_total.labels(
            payment_method=order.payment_method,
            status="success"
        ).inc()
        return result
    except PaymentError:
        orders_total.labels(
            payment_method=order.payment_method,
            status="failed"
        ).inc()
        raise
    finally:
        payment_processing_seconds.observe(time.time() - start)

# Metrics available at GET /metrics

# Install
pip install django-prometheus

# settings.py
INSTALLED_APPS = [
    'django_prometheus',
    # ... other apps
]

MIDDLEWARE = [
    'django_prometheus.middleware.PrometheusBeforeMiddleware',
    # ... your middleware
    'django_prometheus.middleware.PrometheusAfterMiddleware',
]

DATABASES = {
    'default': {
        'ENGINE': 'django_prometheus.db.backends.postgresql',
        # ... connection params
    }
}

# urls.py
from django.urls import include, path

urlpatterns = [
    path('metrics/', include('django_prometheus.urls')),
    # ... other URLs
]

# This automatically provides:
# django_http_requests_total (by method, view, code)
# django_http_request_duration_seconds (by view)
# django_db_query_duration_seconds (by alias, type)
# django_db_query_total
# django_model_inserts_total, django_model_updates_total

# settings.py — log slow queries in development
import logging

LOGGING = {
    'version': 1,
    'handlers': {
        'console': {'class': 'logging.StreamHandler'},
    },
    'loggers': {
        'django.db.backends': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
    },
}

# In production: use django-silk or query-inspector
# django-silk captures request + query timeline in middleware
pip install django-silk

# settings.py (production: only enable for sampling)
MIDDLEWARE = [
    'silk.middleware.SilkyMiddleware',  # Sample mode: SILKY_INTERCEPT_PERCENT
    # ...
]
SILKY_PYTHON_PROFILER = True
SILKY_INTERCEPT_PERCENT = 5  # Profile 5% of requests
SILKY_MAX_REQUEST_BODY_SIZE = -1

# Install core packages
pip install opentelemetry-sdk opentelemetry-exporter-otlp
pip install opentelemetry-instrumentation-django  # or fastapi
pip install opentelemetry-instrumentation-psycopg2  # PostgreSQL
pip install opentelemetry-instrumentation-redis
pip install opentelemetry-instrumentation-celery

# Option 1: Auto-instrumentation (zero code changes)
# Wrap your Gunicorn/uvicorn command:
opentelemetry-instrument   --traces-exporter otlp   --service-name my-django-app   --exporter-otlp-endpoint http://localhost:4317   gunicorn myapp.wsgi:application -w 4

# Option 2: Manual setup in settings.py (more control)
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

provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
)
trace.set_tracer_provider(provider)

# Manual span creation in views
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def process_order(request):
    with tracer.start_as_current_span("process-order") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("customer.tier", customer.tier)
        # ... your business logic

# Celery configuration for monitoring
# celery.py
from celery import Celery
from celery.signals import task_prerun, task_postrun, task_failure
from prometheus_client import Counter, Histogram

app = Celery('myapp')

celery_tasks_total = Counter(
    "celery_tasks_total",
    "Total Celery tasks",
    ["task_name", "status"]
)

celery_task_duration = Histogram(
    "celery_task_duration_seconds",
    "Celery task execution time",
    ["task_name"]
)

@task_prerun.connect
def task_prerun_handler(task_id, task, *args, **kwargs):
    task.start_time = time.time()

@task_postrun.connect
def task_postrun_handler(task_id, task, state, *args, **kwargs):
    duration = time.time() - getattr(task, 'start_time', time.time())
    celery_tasks_total.labels(
        task_name=task.name,
        status=state
    ).inc()
    celery_task_duration.labels(task_name=task.name).observe(duration)

@task_failure.connect
def task_failure_handler(task_id, exception, *args, **kwargs):
    celery_tasks_total.labels(
        task_name=kwargs['sender'].name,
        status="FAILURE"
    ).inc()

# Key metrics to alert on:
# - Queue depth > N (backlog building up)
# - Task failure rate > 5%
# - Task duration p95 > expected SLA