FastAPI Integration

Capture errors and request context from FastAPI applications.

Installation

pip install statly-observe

Setup

from fastapi import FastAPI
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
# Initialize SDK first
Statly.init(
    dsn='https://[email protected]/your-org',
    environment='production',
)
 
app = FastAPI()
 
# Attach to FastAPI app
init_fastapi(app)
 
@app.get('/')
async def hello():
    return {'message': 'Hello World'}
 
@app.get('/error')
async def error():
    raise ValueError('Test error')

What's Captured

Request Context

For each error, the integration captures:

HTTP method and URL
Path and query string
Headers (sanitized)
Path parameters
Query parameters
JSON body (sanitized)
Client IP address

Automatic Sanitization

Sensitive data is filtered from headers and body:

Headers:

authorization
cookie
x-api-key
x-auth-token

Body fields:

password
secret
token
api_key
credit_card
ssn

Manual Capture

Capture errors manually with additional context:

from fastapi import FastAPI
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
app = FastAPI()
init_fastapi(app)
 
@app.get('/users/{user_id}')
async def get_user(user_id: int):
    try:
        user = await fetch_user(user_id)
        return user
    except UserNotFoundError as e:
        Statly.capture_exception(e, context={'user_id': user_id})
        raise

With Authentication

Set user context from authentication:

from fastapi import FastAPI, Depends, Request
from fastapi.security import OAuth2PasswordBearer
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
app = FastAPI()
init_fastapi(app)
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
 
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    Statly.set_user(
        id=str(user.id),
        email=user.email,
    )
    return user
 
@app.get('/me')
async def get_me(user=Depends(get_current_user)):
    return user

Async Support

The integration is fully async-compatible:

from fastapi import FastAPI
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
app = FastAPI()
init_fastapi(app)
 
@app.get('/data')
async def get_data():
    # All async operations are properly handled
    data = await async_fetch_data()
    return data
 
@app.post('/orders')
async def create_order(order: OrderCreate):
    try:
        result = await async_create_order(order)
        return result
    except PaymentError as e:
        # Capture with async context
        Statly.capture_exception(e, context={
            'order_total': order.total,
        })
        raise

Background Tasks

Capture errors in background tasks:

from fastapi import FastAPI, BackgroundTasks
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
app = FastAPI()
init_fastapi(app)
 
def process_order(order_id: str):
    try:
        # Process order
        send_email(order_id)
    except Exception as e:
        Statly.capture_exception(e, context={
            'order_id': order_id,
            'task': 'process_order',
        })
 
@app.post('/orders')
async def create_order(order: OrderCreate, tasks: BackgroundTasks):
    order_id = await save_order(order)
    tasks.add_task(process_order, order_id)
    return {'order_id': order_id}

Exception Handlers

Works with FastAPI exception handlers:

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
app = FastAPI()
init_fastapi(app)
 
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    # Custom errors are still captured
    Statly.capture_exception(exc)
    return JSONResponse(
        status_code=400,
        content={'error': str(exc)},
    )

Full Example

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from statly_observe import Statly
from statly_observe.integrations.fastapi import init_fastapi
 
# Initialize
Statly.init(
    dsn='https://[email protected]/your-org',
    environment='production',
)
 
app = FastAPI(title='My API')
init_fastapi(app)
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
 
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(status_code=401)
    Statly.set_user(id=str(user.id), email=user.email)
    return user
 
@app.get('/api/users')
async def list_users(user=Depends(get_current_user)):
    return await User.find_all()
 
@app.post('/api/orders')
async def create_order(order: OrderCreate, user=Depends(get_current_user)):
    Statly.add_breadcrumb(
        message='Creating order',
        category='order',
        data={'items': len(order.items)},
    )
 
    try:
        result = await process_order(order, user)
        return result
    except PaymentError as e:
        Statly.capture_exception(e)
        raise HTTPException(status_code=400, detail='Payment failed')

Django Installation