FastAPI Middleware Error Recovery: Preventing One Broken Tenant from Taking Down Your Entire SaaS
I learned this lesson the hard way at 2 AM when a customer's misconfigured webhook integration sent 50,000 malformed requests per second to our shared FastAPI origin. One tenant's mistake nearly took down nine others. The middleware caught it, logged it, and the system kept breathing. That's when I realized: middleware isn't about prettifying requests—it's your firewall against cascade failures.
Most FastAPI developers treat middleware as a transformation layer. Request comes in, you extract headers, add context, pass it along. But in multi-tenant systems, middleware is actually your last defense against one tenant's chaos rippling through your entire infrastructure. I'm going to show you exactly how to weaponize it.
The Problem: Why Standard Error Handling Fails in Multi-Tenant Systems
When a request handler throws an exception in a single-tenant app, FastAPI's built-in exception handlers catch it and return a 500. Users see an error. Life goes on. But in multi-tenant systems:
- One tenant's bad data causes a database deadlock → affects all tenants querying that table
- A malformed request body exhausts memory → slows down request processing for everyone
- An uncaught exception in business logic → leaves the connection pool in an unknown state
- Logging doesn't include tenant context → you can't debug whose request caused the crash
I've watched this happen. The culprit wasn't even running expensive queries—it was sending requests so malformed that the validation layer itself crashed. Without tenant-aware middleware, you're blind.
The Solution: Defensive Middleware Layers
The pattern I use now: build a middleware stack that catches exceptions at the boundary, logs tenant context immediately, and returns safe responses without exposing internal state.
Here's the core structure:
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
import logging
import time
import uuid
from typing import Callable
app = FastAPI()
logger = logging.getLogger(__name__)
class TenantContextMiddleware:
"""Extract and validate tenant from request BEFORE it hits handlers."""
def __init__(self, app):
self.app = app
async def __call__(self, request: Request, call_next: Callable):
# Extract tenant ID from subdomain or header
tenant_id = request.headers.get("X-Tenant-ID") or self._extract_from_subdomain(request.url.hostname)
request.state.tenant_id = tenant_id
request.state.request_id = str(uuid.uuid4())
request.state.start_time = time.time()
response = await call_next(request)
return response
def _extract_from_subdomain(self, hostname: str) -> str | None:
if hostname and "." in hostname:
parts = hostname.split(".")
if parts[0] not in ("www", "api", "app"):
return parts[0]
return None
class TenantErrorBoundaryMiddleware:
"""Catch ALL exceptions, log tenant context, return graceful 5xx."""
def __init__(self, app):
self.app = app
async def __call__(self, request: Request, call_next: Callable):
try:
response = await call_next(request)
# Check for 5xx status codes and log them
if 500 <= response.status_code < 600:
self._log_error(request, response.status_code, "Response error")
return response
except Exception as exc:
# This is our last line of defense
self._log_error(request, 500, str(exc), exc)
# Return safe response without stack trace
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"error": "Internal server error",
"request_id": request.state.request_id,
"tenant_id": request.state.tenant_id,
}
)
def _log_error(self, request: Request, status_code: int, message: str, exc: Exception = None):
"""Log with full tenant context for debugging."""
logger.error(
"Request failed",
extra={
"tenant_id": getattr(request.state, "tenant_id", "unknown"),
"request_id": getattr(request.state, "request_id", "unknown"),
"method": request.method,
"path": request.url.path,
"status_code": status_code,
"duration_ms": (time.time() - request.state.start_time) * 1000,
"error": message,
"exception": exc.__class__.__name__ if exc else None,
}
)
# Register in correct order (last registered = first executed)
app.add_middleware(TenantErrorBoundaryMiddleware)
app.add_middleware(TenantContextMiddleware)
This does three things I care about:
- Tenant context extracted early → available throughout the request lifecycle
- Exceptions caught at the boundary → no uncaught exception can crash the app
- Safe error responses → no stack traces leak to clients, but you have request IDs for logs
Rate Limiting and Quota Enforcement: The Next Layer
Now here's where it gets spicy. I add another middleware that actually prevents bad tenants from hammering shared resources:
from redis import Redis
import json
redis_client = Redis(decode_responses=True)
class TenantQuotaMiddleware:
"""Enforce per-tenant rate limits and resource quotas."""
def __init__(self, app, redis=redis_client):
self.app = app
self.redis = redis
async def __call__(self, request: Request, call_next: Callable):
tenant_id = getattr(request.state, "tenant_id", None)
if not tenant_id:
return await call_next(request)
# Check request rate limit
rate_limit_key = f"rate_limit:{tenant_id}"
current_count = self.redis.incr(rate_limit_key)
if current_count == 1:
self.redis.expire(rate_limit_key, 60) # 60 second window
# 1000 requests per minute per tenant (adjust as needed)
if current_count > 1000:
logger.warning(
f"Tenant {tenant_id} exceeded rate limit",
extra={"tenant_id": tenant_id, "count": current_count}
)
return JSONResponse(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
content={
"error": "Rate limit exceeded",
"retry_after": 60,
"request_id": request.state.request_id,
}
)
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(1000 - current_count)
return response
app.add_middleware(TenantQuotaMiddleware)
app.add_middleware(TenantErrorBoundaryMiddleware)
app.add_middleware(TenantContextMiddleware)
The order matters. Tenant context → quota enforcement → error boundary. This ensures you catch rate-limited requests before they even enter your business logic.
What I Missed (and learned from)
Gotcha #1: Middleware order is execution order in reverse. I registered them in logical order once and spent an hour wondering why tenant context wasn't available in the quota middleware. FastAPI executes middleware in reverse registration order—the last one added runs first.
Gotcha #2: Async context managers aren't your friend here. I tried using @asynccontextmanager for cleanup logic (closing connections, rolling back transactions). It worked locally but failed under high load because the context wasn't properly propagated. Stick to explicit try-finally blocks in middleware.
Gotcha #3: Logging context gets lost across async boundaries. I use structlog with context vars now instead of passing state through request objects for log enrichment:
from contextvars import ContextVar
tenant_context: ContextVar[str] = ContextVar("tenant_id", default=None)
class TenantContextMiddleware:
async def __call__(self, request: Request, call_next: Callable):
tenant_id = request.headers.get("X-Tenant-ID")
token = tenant_context.set(tenant_id)
try:
return await call_next(request)
finally:
tenant_context.reset(token)
Now every log across the async chain includes tenant context automatically.
The Real Win
Here's what actually matters: that 2 AM incident? With this middleware stack, the bad tenant got rate-limited in 30 seconds. Their requests returned 429 responses. The database stayed responsive. Other tenants never knew something happened. I got a Slack alert with request_id and tenant_id already in the context, debugged the issue in 10 minutes, and went back to sleep.
That's what defensive middleware buys you in production.
Top comments (0)