FastAPI Request Logging for Multi-Tenant Debugging: Structured Logs That Preserve Tenant Context Without Middleware Chaos
I've spent three weeks digging through 50MB of unstructured text logs trying to figure out why one tenant's Claude API call was poisoning request contexts for everyone else. That week taught me that logging infrastructure is not optional when you're running multi-tenant SaaS at scale—it's the difference between a 30-second fix and a 3-day investigation.
The problem is this: standard FastAPI logging with print statements or basic Python logging spills context everywhere. When tenant A's AI feature call fails, you can't quickly trace which of your 50 concurrent requests it affected. By the time you grep for error messages, you're wading through logs from 12 other tenants. I prefer structured JSON logging with automatic context injection because it makes every log line queryable and tenant-scoped without littering your handler code with context parameters.
Why Unstructured Logging Fails at Multi-Tenant Scale
Most FastAPI projects start with this:
import logging
logger = logging.getLogger(__name__)
@app.post("/api/insights")
async def generate_insights(req: InsightRequest):
logger.info(f"User {req.user_id} requested insights")
# ... API call to Claude
logger.error(f"Claude API failed: {error}")
This burns you the moment you ship to production. Here's why:
- Tenant context is invisible. You log a Claude error but can't immediately see which tenant owns that user. Grep becomes your life.
- Request ID is missing. When one request spawns async tasks, you lose the thread. Which log lines belong together?
-
Context leaks across requests. Python's
contextvarsaren't automatically populated. If you forget to set context in one code path, logs from concurrent requests bleed together. - AI feature context is scattered. You're calling Claude, making database queries, updating caches—but logs don't tell you which operation stage failed or how long each took.
Structured JSON logging solves all of this because:
- Every log line is a queryable object with tenant_id, user_id, request_id baked in.
- You can search by JSON field:
tenant_id:"acme-corp"instead of grepping for company names. - Context is injected automatically—no manual parameter passing through 8 function layers.
- AI feature invocations become traceable events, not scattered print statements.
Architecture: Context Vars + JSON Logging + Middleware
Here's the pattern I use in CitizenApp:
# logging_config.py
import json
import logging
import contextvars
import uuid
from typing import Any, Dict
from datetime import datetime
# Context variables that persist across async function calls
tenant_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"tenant_id", default="unknown"
)
user_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"user_id", default="unknown"
)
request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"request_id", default=str(uuid.uuid4())
)
feature_var: contextvars.ContextVar[str] = contextvars.ContextVar(
"feature", default="unknown"
)
class StructuredLogger(logging.Formatter):
"""Formats logs as JSON with automatic context injection."""
def format(self, record: logging.LogRecord) -> str:
log_data: Dict[str, Any] = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"tenant_id": tenant_id_var.get(),
"user_id": user_id_var.get(),
"request_id": request_id_var.get(),
"feature": feature_var.get(),
}
# Add exception info if present
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
# Allow handlers to inject extra fields
if hasattr(record, "extra_fields"):
log_data.update(record.extra_fields)
return json.dumps(log_data)
def configure_logging():
"""Initialize structured logging."""
handler = logging.StreamHandler()
handler.setFormatter(StructuredLogger())
root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(handler)
return logging.getLogger(__name__)
logger = configure_logging()
Now the middleware that populates context:
# middleware.py
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
import time
import uuid
from logging_config import (
tenant_id_var, user_id_var, request_id_var,
feature_var, logger
)
@app.middleware("http")
async def add_request_context(request: Request, call_next):
"""Extract tenant/user from request and populate context vars."""
request_id = str(uuid.uuid4())
request_id_var.set(request_id)
# Extract tenant from path or header
# In CitizenApp, tenants are identified by subdomain or header
tenant_id = request.headers.get("X-Tenant-ID", "unknown")
tenant_id_var.set(tenant_id)
# Extract user from JWT token or session
user_id = request.headers.get("X-User-ID", "unknown")
user_id_var.set(user_id)
start_time = time.time()
try:
response = await call_next(request)
except Exception as e:
logger.error(f"Request failed: {str(e)}")
raise
finally:
# Log request completion with duration
duration = time.time() - start_time
logger.info(
f"{request.method} {request.url.path} completed",
extra={"extra_fields": {"duration_ms": duration * 1000, "status": response.status_code}}
)
return response
The middleware does one job: extract tenant and user from the request and populate context vars. Everything downstream automatically inherits this context.
Using It in Your Handlers
Now when you log in handlers, context is automatic:
# handlers.py
from fastapi import APIRouter, Depends
from logging_config import logger, feature_var
import anthropic
router = APIRouter()
@router.post("/api/generate-insights")
async def generate_insights(tenant_id: str, user_id: str, req: InsightRequest):
"""
Generate AI insights for a tenant's data.
Every log line is automatically tagged with tenant_id, user_id, request_id.
"""
feature_var.set("insights-generation")
logger.info("Starting insights generation")
try:
# Call Claude API
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
message = await client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[
{
"role": "user",
"content": f"Analyze this data: {req.data}"
}
],
)
logger.info("Claude API call succeeded", extra={
"extra_fields": {"tokens_used": message.usage.output_tokens}
})
# Save results to database
insights = await save_insights(tenant_id, user_id, message.content)
logger.info("Insights saved to database")
return {"insights": insights}
except anthropic.APIError as e:
logger.error(f"Claude API error: {str(e)}")
raise HTTPException(status_code=502, detail="AI service unavailable")
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise
Notice: no manual context passing. No log_context(tenant_id=...) calls everywhere. The logger automatically includes tenant_id, user_id, request_id, and feature because contextvars preserves them across async boundaries.
The output looks like this:
{
"timestamp": "2024-01-15T10:23:45.123456",
"level": "INFO",
"logger": "handlers",
"message": "Starting insights generation",
"tenant_id": "acme-corp",
"user_id": "user-123",
"request_id": "550e8400-e29b-41d4-a716-446655440000",
"feature": "insights-generation"
}
When Claude fails, you immediately see which tenant, which user, which request. If multiple errors occurred, they all share the same request_id, making it trivial to correlate.
Querying Structured Logs
This is where structured logging shines. With Render's log streaming or any centralized logging service (I use Datadog in production), you can query by field:
tenant_id:"acme-corp" AND level:"ERROR"
Versus unstructured grep hell:
grep "acme-corp\|ACME\|error" logs.txt | head -100 | grep -v other_tenants...
Gotcha: Context Vars Don't Persist to Background Tasks
This bit me hard. When you spawn a background task with asyncio.create_task(), the context vars don't automatically propagate:
# WRONG - loses context in background task
asyncio.create_task(process_results_async(results))
# RIGHT - explicitly copy context
import copy_context
ctx = copy_context()
asyncio.create_task(ctx.run(process_results_async, results))
In CitizenApp, I wrap all background task spawning in a helper:
python
def spawn_background_task(coro):
"""Spawn background task preserving context vars."""
ctx = copy_context()
asyncio.create
