FastAPI is the best Python web framework for building APIs in 2024. It combines type-safe request handling via Pydantic, automatic OpenAPI documentation, native async support, and performance that approaches Node.js for I/O-bound workloads. But most FastAPI tutorials stop at “hello world” and never address the challenges you hit at 1,000 requests per second with 50 database connections, long-running background tasks, and a deployment that needs to survive a traffic spike without falling over.
This post covers the specific patterns, configurations, and architectural decisions we use when taking FastAPI applications from prototype to production. Everything here comes from running FastAPI services handling 2,000-10,000 requests per second across multiple client projects over the last two years.
Project Structure That Scales
The flat-file tutorial structure (one main.py with everything) breaks down around 20 endpoints. Routes, database models, Pydantic schemas, utility functions, and middleware all mixed together in a single file becomes unmaintainable and makes testing painful because you cannot import individual components without importing the entire application. Here is the structure we use for every production FastAPI project:
project/
├── app/
│ ├── __init__.py
│ ├── main.py # App factory, middleware, events
│ ├── config.py # Pydantic Settings with env validation
│ ├── dependencies.py # Shared dependency injection
│ ├── middleware.py # Custom middleware (logging, CORS)
│ ├── exceptions.py # Custom exception handlers
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── router.py # Aggregates all v1 routes
│ │ │ ├── products.py
│ │ │ ├── orders.py
│ │ │ └── users.py
│ │ └── health.py # Health check endpoint
│ ├── core/
│ │ ├── database.py # Async SQLAlchemy engine/session
│ │ ├── cache.py # Redis connection pool
│ │ └── security.py # JWT/API key auth utilities
│ ├── models/ # SQLAlchemy ORM models
│ ├── schemas/ # Pydantic request/response schemas
│ ├── services/ # Business logic layer
│ └── tasks/ # Background and scheduled tasks
├── tests/
│ ├── conftest.py # Shared fixtures
│ ├── test_products.py
│ └── test_orders.py
├── alembic/ # Database migrations
├── Dockerfile
├── docker-compose.yml
└── pyproject.tomlThe critical principle: route handlers should be thin. They validate input (via Pydantic schemas), call a service function, and return the result. All business logic lives in the services/ layer. All database operations live in the service layer or dedicated repository functions. Route handlers never import SQLAlchemy directly. This makes testing straightforward because you can test services with a real database and test routes with mocked services, covering both the business logic and the HTTP layer independently.
The config.py module uses Pydantic’s BaseSettings to load and validate configuration from environment variables at startup. This catches misconfiguration immediately rather than at the first request that hits the missing config path:
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
DATABASE_URL: str
REDIS_URL: str = "redis://localhost:6379/0"
JWT_SECRET: str
JWT_ALGORITHM: str = "HS256"
JWT_EXPIRE_MINUTES: int = 30
CORS_ORIGINS: list[str] = ["http://localhost:3000"]
DEBUG: bool = False
LOG_LEVEL: str = "INFO"
model_config = {
"env_file": ".env",
"env_file_encoding": "utf-8"
}
@lru_cache
def get_settings() -> Settings:
return Settings()Database Connection Management
The number one cause of production FastAPI failures we have seen is database connection exhaustion. The default SQLAlchemy async setup without explicit pool configuration creates connections on demand with no upper bound. At 1,000 RPS, that is potentially 1,000 simultaneous database connections if each request takes more than 1 second. PostgreSQL’s default max_connections is 100, so you will hit connection refused errors well before you hit your throughput ceiling.
from sqlalchemy.ext.asyncio import (
create_async_engine, async_sessionmaker, AsyncSession
)
from typing import AsyncGenerator
# Connection pool configuration is critical
engine = create_async_engine(
settings.DATABASE_URL,
pool_size=20, # Steady-state connections
max_overflow=10, # Burst capacity (total max: 30)
pool_timeout=30, # Seconds to wait for a connection
pool_recycle=3600, # Recycle connections after 1 hour
pool_pre_ping=True, # Verify connections before use
echo=False, # NEVER True in production
connect_args={
"server_settings": {
"statement_timeout": "30000", # 30s query timeout
"idle_in_transaction_session_timeout": "60000"
}
}
)
SessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False # Prevent lazy loading after commit
)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with SessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raisePool sizing math
The formula we use: total pool connections across all workers should not exceed your PostgreSQL max_connections minus 10 (reserved for admin, monitoring, and migration connections). With 4 Uvicorn workers, pool_size=20 and max_overflow=10 means up to 30 connections per worker, 120 total across all workers. Set PostgreSQL’s max_connections to at least 130.
If you need more application connections than PostgreSQL can handle (common with microservices where 5+ services each need connection pools), put PgBouncer in front. PgBouncer pools connections at the proxy level, multiplexing many application connections onto fewer PostgreSQL connections. In transaction mode, 200 application connections can share 30 PostgreSQL connections. This is especially important on managed PostgreSQL services (RDS, Cloud SQL) where max_connections is limited by the instance size and cannot be increased without upgrading.
The pool_pre_ping=True setting is important for long-running applications. Without it, connections that have been idle for longer than PostgreSQL’s tcp_keepalive_idle timeout (or that were terminated by a firewall, load balancer, or PostgreSQL restart) remain in the pool as dead connections. The next request that gets a dead connection will fail. With pre-ping enabled, SQLAlchemy sends a lightweight query (SELECT 1) before handing the connection to the application, detecting and discarding dead connections automatically.
Response Caching with Redis
The fastest database query is the one you do not make. For read-heavy APIs (which is most APIs: typical read-to-write ratios are 10:1 to 100:1), Redis caching can reduce database load by 70-90%. We implement caching as a decorator that can be applied to any async endpoint:
import redis.asyncio as redis
import hashlib
import json
from functools import wraps
from typing import Callable
redis_pool = redis.ConnectionPool.from_url(
settings.REDIS_URL,
max_connections=50,
decode_responses=True
)
redis_client = redis.Redis(connection_pool=redis_pool)
def cache_response(
ttl: int = 300,
key_prefix: str = "",
vary_on: list[str] | None = None
):
"""Cache endpoint responses in Redis.
Args:
ttl: Cache TTL in seconds
key_prefix: Prefix for cache keys
vary_on: Header names to include in cache key
"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
# Build cache key from function, args, and headers
key_parts = [
key_prefix or func.__name__,
str(sorted(kwargs.items()))
]
key_data = ":".join(key_parts)
cache_key = f"cache:{hashlib.sha256(key_data.encode()).hexdigest()[:16]}"
# Try cache first
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Cache miss: call the actual function
result = await func(*args, **kwargs)
# Serialize and store
serialized = json.dumps(
result, default=str
)
await redis_client.set(
cache_key, serialized, ex=ttl
)
return result
return wrapper
return decorator
# Usage on an endpoint
@router.get("/products/{product_id}")
@cache_response(ttl=600, key_prefix="product")
async def get_product(
product_id: int,
db: AsyncSession = Depends(get_db)
):
product = await product_service.get_by_id(
db, product_id
)
if not product:
raise HTTPException(status_code=404)
return ProductResponse.model_validate(product)Cache invalidation is the hard part, as the saying goes. We use three strategies depending on the data’s characteristics. For product data that changes infrequently, we use time-based expiration (10-minute TTL) and accept that updates take up to 10 minutes to appear. For user-specific data, we use explicit invalidation: when a user updates their profile, we delete all cache keys with the user:{user_id}: prefix. For high-frequency data like search results, we use short TTLs (30-60 seconds) as a simple traffic absorber that prevents identical concurrent queries from all hitting the database.
Rate Limiting and Request Throttling
Every production API needs rate limiting. Without it, a single misbehaving client (or a bot, or an attacker) can consume all your server resources and degrade the experience for everyone. We use a sliding window rate limiter backed by Redis, implemented as middleware that runs before any route handler:
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(
self, app,
requests_per_minute: int = 60,
burst_limit: int = 10
):
super().__init__(app)
self.rpm = requests_per_minute
self.burst = burst_limit
async def dispatch(self, request: Request, call_next):
# Skip rate limiting for health checks
if request.url.path == "/health":
return await call_next(request)
# Identify client by API key (preferred) or IP
client_id = (
request.headers.get('X-API-Key')
or request.client.host
)
key = f"ratelimit:{client_id}"
now = time.time()
window_start = now - 60
pipe = redis_client.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zcard(key)
pipe.zadd(key, {str(now): now})
pipe.expire(key, 120)
results = await pipe.execute()
request_count = results[1]
remaining = max(0, self.rpm - request_count - 1)
if request_count >= self.rpm:
return JSONResponse(
status_code=429,
content={"detail": "Rate limit exceeded"},
headers={
'Retry-After': '60',
'X-RateLimit-Limit': str(self.rpm),
'X-RateLimit-Remaining': '0',
'X-RateLimit-Reset': str(int(now) + 60)
}
)
response = await call_next(request)
response.headers['X-RateLimit-Limit'] = str(self.rpm)
response.headers['X-RateLimit-Remaining'] = str(remaining)
return responseThe sliding window approach (using Redis sorted sets with timestamps as scores) is more accurate than the fixed window approach. A fixed window of 60 requests per minute allows 120 requests in 2 seconds if they straddle the window boundary (60 at the end of one window, 60 at the start of the next). The sliding window correctly limits to 60 requests in any 60-second period. The Redis sorted set operations are atomic and take less than 0.1ms per request, so the rate limiting overhead is negligible.
Background Tasks and Job Queues
FastAPI’s built-in BackgroundTasks is fine for fire-and-forget work that can be lost if the process crashes (logging, analytics events, non-critical notifications). For anything important that must complete reliably (sending confirmation emails, processing file uploads, syncing data to third-party services), use a proper job queue. We use ARQ (async Redis queue) for most projects because it is lightweight, async-native, and uses Redis that we already have for caching:
from arq import create_pool, cron
from arq.connections import RedisSettings
# Define worker functions
async def process_upload(ctx: dict, upload_id: int):
"""Process an uploaded file (resize images, extract text, etc.)"""
async with get_db_session() as db:
upload = await upload_service.get(db, upload_id)
if not upload:
return # Already deleted, skip
try:
result = await heavy_processing(upload)
await upload_service.mark_complete(
db, upload_id, result=result
)
except Exception as e:
await upload_service.mark_failed(
db, upload_id, error=str(e)
)
raise # ARQ will retry based on config
async def send_welcome_email(ctx: dict, user_id: int):
"""Send welcome email to new user."""
async with get_db_session() as db:
user = await user_service.get(db, user_id)
await email_service.send_template(
to=user.email,
template="welcome",
context={"name": user.name}
)
# Worker class configuration
class WorkerSettings:
functions = [process_upload, send_welcome_email]
redis_settings = RedisSettings.from_dsn(settings.REDIS_URL)
max_jobs = 10
job_timeout = 300 # 5 minutes max per job
retry_jobs = True
max_tries = 3
health_check_interval = 30
# Enqueue from API handler
@router.post("/uploads", status_code=202)
async def create_upload(
file: UploadFile,
db: AsyncSession = Depends(get_db),
queue: ArqRedis = Depends(get_queue)
):
upload = await upload_service.create(db, file)
await queue.enqueue_job(
'process_upload', upload.id
)
return {
"upload_id": upload.id,
"status": "processing",
"status_url": f"/uploads/{upload.id}/status"
}For heavier workloads or teams that need Celery’s ecosystem (scheduled tasks, task chains, Canvas workflows), Celery with Redis as the broker works well. The main advantage of ARQ over Celery for FastAPI projects is that ARQ is async-native (no sync-to-async bridging needed), lighter weight (no separate beat process for scheduling), and uses the same Redis you already have. The main advantage of Celery is its maturity, larger ecosystem of extensions, and better monitoring tools (Flower dashboard).
Deployment: Uvicorn + Gunicorn + Docker
The production deployment stack is Gunicorn as the process manager, Uvicorn workers for async request handling, Docker for containerization, and Nginx as the reverse proxy for TLS termination, static files, and request buffering:
# Dockerfile
FROM python:3.12-slim AS base
# Install system deps in a separate layer for caching
RUN apt-get update && apt-get install -y --no-install-recommends
libpq-dev && rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Install Python deps in a separate layer
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Non-root user for security
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
# Health check
HEALTHCHECK --interval=30s --timeout=5s --retries=3
CMD python -c "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"
# Gunicorn with Uvicorn workers
CMD ["gunicorn", "app.main:app",
"-w", "4",
"-k", "uvicorn.workers.UvicornWorker",
"--bind", "0.0.0.0:8000",
"--timeout", "120",
"--graceful-timeout", "30",
"--max-requests", "10000",
"--max-requests-jitter", "1000",
"--access-logfile", "-",
"--error-logfile", "-"]The --max-requests flag is critically important. It recycles workers after 10,000 requests (with jitter of up to 1,000 to prevent all workers from restarting simultaneously). This prevents memory leaks from accumulating over time. Without it, we have seen FastAPI workers grow from 200MB to 2GB over 48 hours due to memory leaks in third-party libraries (SQLAlchemy connection objects not being properly garbage collected, httpx response bodies accumulating in debug mode, and Pydantic model caches growing without bound). The jitter ensures that at most one worker is restarting at any given time, so the other 3 workers continue serving requests.
Worker count follows the formula: workers = (2 * CPU_cores) + 1 for I/O-bound applications. For a 2-core server, that is 5 workers. For a 4-core server, 9 workers. If your application is CPU-bound (ML inference in the request path, heavy computation), reduce to CPU_cores + 1. We typically use 4 workers on a 2-core instance because the +1 formula is a guideline, not a rule, and 4 is easier to reason about for pool sizing math.
Observability: The Non-Negotiables
Every production FastAPI service must have these four observability layers. Skip any one and you are flying blind when something goes wrong at 2 AM:
- Structured logging with
structlogorpython-json-logger. Every log line must include request_id (for tracing a request across log entries), user_id (for investigating user-reported issues), endpoint path, response status code, and duration_ms. JSON format for machine parsing by your log aggregator (ELK, Loki, Datadog). Never use print statements or unstructured logging in production. - Metrics via Prometheus client library. Expose: request count by endpoint and status code (counter), request latency histogram by endpoint (histogram with buckets at 10ms, 50ms, 100ms, 250ms, 500ms, 1s, 5s), active in-flight requests (gauge), database connection pool utilization (gauge: active/total), Redis cache hit rate (counter for hits and misses), and background task queue depth (gauge).
- Health check endpoint that verifies all critical dependencies.
- Distributed tracing with OpenTelemetry. Propagate trace IDs through HTTP headers, database queries, Redis operations, and background task enqueues so you can trace a single user request through every system it touches.
@router.get("/health")
async def health_check(
db: AsyncSession = Depends(get_db)
):
checks = {}
overall_healthy = True
# Database connectivity
try:
result = await db.execute(text("SELECT 1"))
checks["database"] = {"status": "healthy"}
except Exception as e:
checks["database"] = {
"status": "unhealthy", "error": str(e)
}
overall_healthy = False
# Redis connectivity
try:
await redis_client.ping()
info = await redis_client.info("memory")
checks["redis"] = {
"status": "healthy",
"used_memory_mb": round(
info["used_memory"] / 1024 / 1024, 1
)
}
except Exception as e:
checks["redis"] = {
"status": "unhealthy", "error": str(e)
}
overall_healthy = False
# Database pool stats
pool = engine.pool
checks["db_pool"] = {
"size": pool.size(),
"checked_out": pool.checkedout(),
"overflow": pool.overflow(),
"checked_in": pool.checkedin()
}
return JSONResponse(
status_code=200 if overall_healthy else 503,
content={
"status": "ok" if overall_healthy else "degraded",
"checks": checks,
"version": settings.APP_VERSION
}
)FastAPI is an excellent foundation for production APIs. But the framework gives you the building blocks, not the building. The patterns above (connection pooling, caching, rate limiting, structured background tasks, proper deployment configuration, and comprehensive observability) are the difference between a prototype that works on your laptop and a service that handles real traffic reliably under real-world conditions.
