Python Sentry Integration: Error Tracking and Performance

Sentry is the most widely used error tracking and performance monitoring platform for Python applications. The Sentry SDK auto-instruments FastAPI, Django, Celery, SQLAlchemy, httpx, and Redis — capturing exceptions with full stack traces, request context, breadcrumbs, and performance traces with zero configuration beyond sentry_sdk.init(). This guide covers setup, performance monitoring, user context, custom events, and filtering noise.

Installation and Basic Setup

pip install "sentry-sdk[fastapi,sqlalchemy,celery,httpx,redis]"

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
from sentry_sdk.integrations.celery import CeleryIntegration
from sentry_sdk.integrations.httpx import HttpxIntegration
from sentry_sdk.integrations.redis import RedisIntegration
import os

sentry_sdk.init(
    dsn=os.environ["SENTRY_DSN"],
    environment=os.environ.get("ENVIRONMENT", "production"),
    release=os.environ.get("GIT_SHA", "unknown"),
    traces_sample_rate=0.1,      # 10% of transactions traced
    profiles_sample_rate=0.05,   # 5% profiled (requires traces)
    send_default_pii=False,      # GDPR: don't send IP, cookies, user data by default
    integrations=[
        FastApiIntegration(transaction_style="endpoint"),
        SqlalchemyIntegration(),
        CeleryIntegration(),
        HttpxIntegration(),
        RedisIntegration(),
    ],
    before_send=filter_events,   # see Filtering section
)

FastAPI Integration

The FastAPI integration automatically captures unhandled exceptions, creates performance transactions for each request, and attaches request data (URL, headers, query params) as event context. You get request-level tracing and error grouping out of the box.

from contextlib import asynccontextmanager
import sentry_sdk
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


@asynccontextmanager
async def lifespan(app: FastAPI):
    sentry_sdk.init(
        dsn=os.environ["SENTRY_DSN"],
        integrations=[FastApiIntegration()],
        traces_sample_rate=0.1,
    )
    yield


app = FastAPI(lifespan=lifespan)


# Sentry automatically captures this — no extra code needed
@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    if order_id < 0:
        raise ValueError(f"Invalid order_id: {order_id}")
    return {"order_id": order_id}


# Custom error handler that still reports to Sentry
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Sentry integration already captured this; we just return a response
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error", "request_id": request.headers.get("X-Request-ID")},
    )


# Capture a handled exception (not raised, but still worth tracking)
@app.post("/payments")
async def process_payment(data: dict):
    try:
        result = await charge_card(data)
        return result
    except PaymentDeclinedError as e:
        # Don't raise — it's expected — but report to Sentry for monitoring
        sentry_sdk.capture_exception(e)
        return JSONResponse(status_code=402, content={"error": "payment_declined"})


async def charge_card(data): ...
class PaymentDeclinedError(Exception): ...

Performance Monitoring

Sentry Performance shows which endpoints are slow, which database queries are N+1, and where time is spent across your stack. The SQL integration annotates each query with duration, and Sentry's query insights surface repeated slow queries automatically.

import sentry_sdk
from sentry_sdk import start_transaction


# Manual performance transaction (for non-HTTP code paths)
def generate_report(report_id: str) -> dict:
    with sentry_sdk.start_transaction(op="task", name=f"generate_report/{report_id}") as txn:
        txn.set_tag("report.id", report_id)

        with sentry_sdk.start_span(op="db.query", description="fetch_report_data"):
            data = fetch_report_data(report_id)    # auto-traced via SQLAlchemy integration

        with sentry_sdk.start_span(op="compute", description="aggregate_metrics"):
            metrics = aggregate(data)

        with sentry_sdk.start_span(op="render", description="build_pdf"):
            pdf = render_pdf(metrics)

        txn.set_measurement("report.rows", len(data), unit="row")
        txn.set_measurement("report.size_kb", len(pdf) / 1024, unit="kilobyte")
        return {"pdf_size": len(pdf)}


def fetch_report_data(report_id): return []
def aggregate(data): return {}
def render_pdf(metrics): return b""


# Custom sampling — trace all slow requests, sample fast ones
def traces_sampler(sampling_context: dict) -> float:
    # Always trace admin endpoints
    if sampling_context.get("wsgi_environ", {}).get("PATH_INFO", "").startswith("/admin"):
        return 1.0
    # Always trace errors (handled separately by Sentry)
    # 10% sample for normal traffic
    return 0.1

User Context and Scope

Attach user identity to Sentry events so you can filter by user in the Issues UI, build impact alerts ("this error affects 500 users"), and comply with GDPR (only attach non-PII identifiers).

import sentry_sdk
from fastapi import FastAPI, Request, Depends
from sentry_sdk import configure_scope, push_scope


# Set user on every authenticated request
async def get_current_user(request: Request) -> dict:
    # ... validate JWT token ...
    return {"id": 42, "email": "user@example.com", "role": "admin"}


async def set_sentry_user(user: dict = Depends(get_current_user)):
    with configure_scope() as scope:
        scope.set_user({
            "id": str(user["id"]),    # string ID — never email for GDPR
            # "email": user["email"],  # omit if send_default_pii=False
            "username": f"user-{user['id']}",
        })
        scope.set_tag("user.role", user["role"])
    return user


# Tags and extra context
def set_request_context(request: Request, user: dict):
    with configure_scope() as scope:
        scope.set_tag("api.version", request.headers.get("X-API-Version", "v1"))
        scope.set_tag("tenant", request.headers.get("X-Tenant-ID", "default"))
        scope.set_extra("request_id", request.headers.get("X-Request-ID"))
        scope.set_context("order", {"id": "ord-123", "status": "processing"})


# Isolated scope for a single operation
def process_webhook(payload: dict):
    with push_scope() as scope:
        scope.set_tag("webhook.type", payload.get("event"))
        scope.set_extra("webhook_payload", payload)
        try:
            handle_webhook(payload)
        except Exception as e:
            sentry_sdk.capture_exception(e)
            raise


def handle_webhook(payload): ...

Custom Events and Breadcrumbs

Capture custom messages, create manual breadcrumbs to trace the path leading to an error, and send custom events for business-critical flows you want to monitor regardless of exceptions.

import sentry_sdk


# Capture a message (non-exception event)
sentry_sdk.capture_message("Payment gateway timeout exceeded SLA", level="warning")

# Add breadcrumbs — show what happened before an error
def process_order(order_id: str, user_id: int):
    sentry_sdk.add_breadcrumb(
        category="order",
        message=f"Starting order processing for {order_id}",
        level="info",
        data={"order_id": order_id, "user_id": user_id},
    )

    try:
        inventory_ok = check_inventory(order_id)
        sentry_sdk.add_breadcrumb(
            category="inventory",
            message=f"Inventory check {'passed' if inventory_ok else 'failed'}",
            level="info" if inventory_ok else "warning",
            data={"available": inventory_ok},
        )

        charge_id = charge_customer(user_id)
        sentry_sdk.add_breadcrumb(
            category="payment",
            message=f"Payment charged: {charge_id}",
            level="info",
        )

        fulfill_order(order_id)

    except Exception as e:
        # Breadcrumbs appear in Sentry alongside the exception
        sentry_sdk.capture_exception(e)
        raise


def check_inventory(order_id): return True
def charge_customer(user_id): return "ch_123"
def fulfill_order(order_id): ...

Celery Integration

The Celery integration captures task failures, retries, and performance data. Task failures appear in Sentry with the full exception chain and task arguments as context. Configure it before defining tasks.

import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration
from celery import Celery

# Init Sentry BEFORE creating the Celery app
sentry_sdk.init(
    dsn=os.environ["SENTRY_DSN"],
    integrations=[CeleryIntegration(monitor_beat_tasks=True)],
    traces_sample_rate=0.05,
)

celery_app = Celery("tasks", broker="redis://localhost:6379/0")


@celery_app.task(bind=True, max_retries=3)
def process_payment(self, order_id: str, amount: float):
    try:
        result = charge_stripe(order_id, amount)
        return result
    except TransientError as e:
        # Sentry captures this on final retry exhaustion
        raise self.retry(exc=e, countdown=60)
    except PermanentError as e:
        # Sentry captures this immediately — no retry
        sentry_sdk.capture_exception(e)
        raise


class TransientError(Exception): ...
class PermanentError(Exception): ...
def charge_stripe(order_id, amount): ...

Filtering Noise and PII Scrubbing

Use before_send to drop non-actionable errors and scrub PII before events leave your server. This keeps your Sentry quota clean and ensures GDPR compliance.

import sentry_sdk
from sentry_sdk.types import Event, Hint


IGNORED_ERRORS = {
    "ConnectionResetError",
    "BrokenPipeError",
    "asyncio.CancelledError",
}

SENSITIVE_KEYS = {"password", "token", "secret", "api_key", "authorization", "credit_card"}


def filter_events(event: Event, hint: Hint) -> Event | None:
    # Drop known non-actionable errors
    exception = event.get("exception", {}).get("values", [{}])[0]
    if exception.get("type") in IGNORED_ERRORS:
        return None

    # Drop health check 404s
    request = event.get("request", {})
    if request.get("url", "").endswith(("/health", "/ping")):
        return None

    # Scrub sensitive fields from request data
    if "data" in request:
        for key in list(request["data"].keys()):
            if key.lower() in SENSITIVE_KEYS:
                request["data"][key] = "[REDACTED]"

    return event


sentry_sdk.init(
    dsn=os.environ["SENTRY_DSN"],
    before_send=filter_events,
    ignore_errors=[KeyboardInterrupt, SystemExit],
)

Frequently Asked Questions

What traces_sample_rate should I use in production?

Start with 0.1 (10%) for moderate traffic. For very high traffic (>1000 req/s), use 0.01–0.05 to stay within quota. Use a traces_sampler function to always trace admin routes, errors, and slow requests while sampling normal traffic.

How do I avoid sending PII to Sentry?

Set send_default_pii=False (default). Use before_send to scrub additional fields. Never attach email addresses to user context — use user IDs instead. Enable Sentry's Data Scrubbing rules in the project settings as a second layer of defence.

Does Sentry SDK add latency?

Negligible. The SDK batches events and sends them asynchronously in a background thread. Instrumentation overhead is <1ms per request. Performance traces add a small amount of context collection but are sampled, so most requests have zero Sentry overhead.