Skip to main content

FastAPI Integration

The FastAPI integration provides automatic error capture and request context for all endpoints.

Installation

First, install with FastAPI support:
pip install proliferate[fastapi]

Setup with Middleware

1

Initialize the SDK

Initialize Proliferate early in your application startup, before creating the FastAPI app:
import proliferate

proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)
2

Add Middleware

Add ProliferateMiddleware to your FastAPI app:
from fastapi import FastAPI
from proliferate.integrations.fastapi import ProliferateMiddleware

app = FastAPI()
app.add_middleware(ProliferateMiddleware)
3

Add Context in Routes

Set user and account context in your route handlers:
@app.post("/api/checkout")
async def checkout(request: CheckoutRequest, user: User = Depends(get_user)):
    # Set context for this request
    proliferate.set_user({"id": str(user.id), "email": user.email})
    proliferate.set_account({"id": str(user.account_id), "name": user.account_name})
    proliferate.set_tag("endpoint", "checkout")

    # Your business logic here
    # Any unhandled exception will be automatically captured with full context
    result = await process_checkout(request)
    return result

Complete FastAPI Example

main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
import proliferate
from proliferate.integrations.fastapi import ProliferateMiddleware

# Initialize SDK before creating app
proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)

app = FastAPI(title="My API")

# Add Proliferate middleware
app.add_middleware(ProliferateMiddleware)

# Sample models
class User(BaseModel):
    id: str
    email: str
    account_id: str
    account_name: str

class CheckoutRequest(BaseModel):
    items: list[str]
    total: float

# Sample dependency
async def get_current_user() -> User:
    # Your authentication logic here
    return User(
        id="user_123",
        email="[email protected]",
        account_id="acct_456",
        account_name="Acme Corp"
    )

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.post("/api/checkout")
async def checkout(
    request: CheckoutRequest,
    user: User = Depends(get_current_user)
):
    # Set context for error reporting
    proliferate.set_user({"id": user.id, "email": user.email})
    proliferate.set_account({"id": user.account_id, "name": user.account_name})
    proliferate.set_tag("checkout_amount", request.total)

    # Simulate an error
    if request.total > 10000:
        raise ValueError("Order amount exceeds limit")

    return {"status": "success", "order_id": "order_789"}

@app.get("/api/users/{user_id}")
async def get_user(user_id: str, current_user: User = Depends(get_current_user)):
    # Set context
    proliferate.set_user({"id": current_user.id, "email": current_user.email})

    # Manual error capture for handled exceptions
    try:
        user_data = await fetch_user(user_id)
        return user_data
    except UserNotFoundError as e:
        # Capture but don't re-raise
        proliferate.capture_exception(e, extra={"requested_user_id": user_id})
        raise HTTPException(status_code=404, detail="User not found")

What’s Captured Automatically

The FastAPI middleware automatically captures:
  • Method: HTTP method (GET, POST, etc.)
  • URL: Full request URL including query parameters
  • Path: Request path (e.g., /api/checkout)
  • Headers: Request headers (sensitive headers filtered)
  • Query Parameters: Parsed query string parameters
  • Client IP: Original client IP (handles X-Forwarded-For proxy headers)
  • Request ID: Generated UUID or from X-Request-ID / X-Correlation-ID header
These headers are automatically filtered from error reports:
  • authorization
  • cookie
  • set-cookie
  • x-api-key
  • api-key
  • x-auth-token
  • x-csrf-token
  • x-xsrf-token
  • Exception Type: Full module path (e.g., stripe.error.CardError)
  • Message: Exception message
  • Stack Trace: Complete Python traceback
  • Timestamp: UTC timestamp of the error

Flask Integration

The Flask integration provides the same automatic error capture for Flask applications.

Setup

1

Initialize the SDK

import proliferate

proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)
2

Initialize Flask Integration

from flask import Flask
from proliferate.integrations.flask import init_app

app = Flask(__name__)
init_app(app)
3

Add Context in Routes

@app.route("/api/checkout", methods=["POST"])
def checkout():
    user = get_current_user()

    proliferate.set_user({"id": user.id, "email": user.email})
    proliferate.set_account({"id": user.account_id, "name": user.account_name})

    # Your business logic
    result = process_checkout(request.json)
    return jsonify(result)

Complete Flask Example

app.py
from flask import Flask, request, jsonify
import proliferate
from proliferate.integrations.flask import init_app

# Initialize SDK
proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)

app = Flask(__name__)

# Initialize Flask integration
init_app(app)

# Mock user data for demo
def get_current_user():
    return {
        "id": "user_123",
        "email": "[email protected]",
        "account_id": "acct_456",
        "account_name": "Acme Corp"
    }

@app.route("/")
def root():
    return jsonify({"message": "Hello World"})

@app.route("/api/checkout", methods=["POST"])
def checkout():
    user = get_current_user()

    # Set context for this request
    proliferate.set_user({"id": user["id"], "email": user["email"]})
    proliferate.set_account({"id": user["account_id"], "name": user["account_name"]})

    data = request.json
    total = data.get("total", 0)

    proliferate.set_tag("checkout_amount", total)

    # Simulate error
    if total > 10000:
        raise ValueError("Order amount exceeds limit")

    return jsonify({"status": "success", "order_id": "order_789"})

@app.route("/api/users/<user_id>")
def get_user(user_id):
    current_user = get_current_user()
    proliferate.set_user({"id": current_user["id"], "email": current_user["email"]})

    # Manual error capture for handled exceptions
    try:
        user_data = fetch_user(user_id)
        return jsonify(user_data)
    except UserNotFoundError as e:
        proliferate.capture_exception(e, extra={"requested_user_id": user_id})
        return jsonify({"error": "User not found"}), 404

if __name__ == "__main__":
    app.run(debug=True)

Alternative: Class-based Flask Extension

For the Flask application factory pattern:
from flask import Flask
from proliferate.integrations.flask import FlaskProliferate

# Create extension instance
proliferate_ext = FlaskProliferate()

def create_app():
    app = Flask(__name__)

    # Initialize SDK
    import proliferate
    proliferate.init(
        endpoint="https://api.proliferate.com/api/v1/errors",
        api_key="pk_your_project_key",
    )

    # Initialize extension
    proliferate_ext.init_app(app)

    return app

Other ASGI/WSGI Frameworks

For frameworks not directly supported, you can manually capture errors:

Django Example

settings.py
import proliferate

proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)
middleware.py
from proliferate import capture_exception, set_user

class ProliferateMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Set user context if authenticated
        if request.user.is_authenticated:
            set_user({
                "id": str(request.user.id),
                "email": request.user.email
            })

        response = self.get_response(request)
        return response

    def process_exception(self, request, exception):
        # Capture the exception
        capture_exception(exception)
        return None  # Let Django handle the response

Tornado Example

import tornado.web
import proliferate

proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_your_project_key",
)

class BaseHandler(tornado.web.RequestHandler):
    def write_error(self, status_code, **kwargs):
        # Capture exceptions
        if "exc_info" in kwargs:
            import sys
            exc_info = kwargs["exc_info"]
            proliferate.capture_exception(exc_info[1])

        super().write_error(status_code, **kwargs)

Context Isolation

The Python SDK uses contextvars for context storage, which provides automatic isolation for:
  • Async tasks (asyncio)
  • Concurrent requests in ASGI apps (FastAPI, Starlette)
  • Threads
Each request/task gets its own isolated context that won’t leak to other concurrent requests.

Testing

Disable error sending during tests:
# In your test setup
proliferate.init(
    endpoint="https://api.proliferate.com/api/v1/errors",
    api_key="pk_test_key",
    enabled=False,  # Don't send errors during tests
)

Next Steps