FastAPI Production Checklist
When moving your FastAPI application from development to production, there are several critical considerations to ensure your application is secure, performant, and reliable. This checklist will help you prepare your FastAPI application for production deployment.
Introduction
Building a FastAPI application for development is one thing, but deploying it to production requires additional considerations. Production environments face real users, potential security threats, varying loads, and need proper monitoring and maintenance.
This guide covers the essential steps to ensure your FastAPI application is production-ready, focusing on:
- Security configurations
- Performance optimizations
- Deployment strategies
- Monitoring and logging
- Error handling
- Documentation
Security Considerations
1. Secure CORS Configuration
Cross-Origin Resource Sharing (CORS) should be properly configured to prevent unauthorized domains from accessing your API.
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# Set allowed origins explicitly instead of allowing all with "*"
app.add_middleware(
CORSMiddleware,
allow_origins=["https://yourdomain.com", "https://www.yourdomain.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
2. Implement Authentication & Authorization
Every production API should have proper authentication and authorization:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional
# Setup OAuth2 with Password flow
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "YOUR_SECURE_SECRET_KEY" # Store in environment variables
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
return username
except JWTError:
raise credentials_exception
@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_user)):
return {"username": current_user}
3. Use Environment Variables for Secrets
Never hard-code sensitive information like database credentials, API keys, or JWT secrets:
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Access environment variables
DATABASE_URL = os.getenv("DATABASE_URL")
API_KEY = os.getenv("API_KEY")
SECRET_KEY = os.getenv("SECRET_KEY")
4. Enable HTTPS
Always use HTTPS in production. Configure your web server (Nginx, Apache) to handle SSL/TLS termination.
Example Nginx configuration snippet:
server {
listen 443 ssl;
server_name api.yourdomain.com;
ssl_certificate /path/to/certificate.crt;
ssl_certificate_key /path/to/private.key;
location / {
proxy_pass http://localhost:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
5. Implement Rate Limiting
Protect your API from abuse by implementing rate limiting:
from fastapi import FastAPI, Request
from fastapi.middleware import Middleware
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/")
@limiter.limit("5/minute")
async def root(request: Request):
return {"message": "Hello World"}
Performance Optimization
1. Use Async Database Connections
For database operations, use asynchronous clients to maximize FastAPI's asynchronous capabilities:
from fastapi import FastAPI, Depends
import databases
import sqlalchemy
from typing import List
from pydantic import BaseModel
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
notes = sqlalchemy.Table(
"notes",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("text", sqlalchemy.String),
sqlalchemy.Column("completed", sqlalchemy.Boolean),
)
class NoteIn(BaseModel):
text: str
completed: bool
class Note(BaseModel):
id: int
text: str
completed: bool
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
@app.get("/notes/", response_model=List[Note])
async def read_notes():
query = notes.select()
return await database.fetch_all(query)
2. Implement Caching
For frequently accessed, rarely changing data, implement caching:
from fastapi import FastAPI, Response
import redis
import json
import pickle
from datetime import timedelta
app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)
@app.get("/product/{product_id}")
async def get_product(product_id: int, response: Response):
cache_key = f"product:{product_id}"
# Try to get from cache
cached_product = redis_client.get(cache_key)
if cached_product:
response.headers["X-Cache"] = "HIT"
return pickle.loads(cached_product)
# If not in cache, get from database (simulated here)
product = await get_product_from_db(product_id)
# Store in cache for 1 hour
redis_client.setex(
cache_key,
timedelta(hours=1),
pickle.dumps(product)
)
response.headers["X-Cache"] = "MISS"
return product
async def get_product_from_db(product_id: int):
# Simulate database fetch
return {"id": product_id, "name": f"Product {product_id}", "price": 29.99}
3. Use Connection Pooling
For database connections, implement connection pooling to reuse connections:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
# Create engine with a connection pool
engine = create_async_engine(
DATABASE_URL,
pool_size=20, # Maximum number of connections in the pool
max_overflow=10, # Maximum number of connections that can be created beyond pool_size
pool_timeout=30, # Seconds to wait before giving up on getting a connection
pool_recycle=1800, # Recycle connections after 30 minutes
)
# Create session factory
AsyncSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()
Deployment Strategies
1. Use a Production ASGI Server
FastAPI uses Uvicorn for development, but for production, you should use Gunicorn with Uvicorn workers:
# Install required packages
pip install gunicorn uvicorn
# Run in production with multiple workers
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
You can create a start script (start.sh
):
#!/bin/bash
# Calculate workers based on CPU cores
WORKERS=$(( 2 * $(nproc) + 1 ))
# Start Gunicorn with Uvicorn workers
exec gunicorn -w $WORKERS \
-k uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--log-level=info \
--access-logfile=- \
--error-logfile=- \
main:app
2. Containerize Your Application
Create a Dockerfile for your FastAPI application:
FROM python:3.9-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Run with Gunicorn and Uvicorn workers
CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "main:app", "--bind", "0.0.0.0:8000"]
3. Set Up Health Checks
Implement health check endpoints for monitoring and container orchestration:
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
@app.get("/health")
async def health_check():
return {"status": "healthy"}
@app.get("/health/db", status_code=200)
async def db_health_check(db: AsyncSession = Depends(get_db)):
try:
# Execute a simple query
result = await db.execute("SELECT 1")
if result:
return {"status": "database healthy"}
except Exception as e:
return {"status": "database unhealthy", "error": str(e)}
Monitoring and Logging
1. Configure Comprehensive Logging
Set up proper logging for your application:
import logging
from fastapi import FastAPI, Request
import time
from typing import Callable
import uvicorn
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next: Callable):
start_time = time.time()
# Log request details
logger.info(f"Request started: {request.method} {request.url.path}")
response = await call_next(request)
# Log response details
process_time = time.time() - start_time
logger.info(f"Request completed: {request.method} {request.url.path} - Status: {response.status_code} - Time: {process_time:.4f}s")
return response
@app.get("/")
async def root():
logger.info("Root endpoint called")
return {"message": "Hello World"}
# Handle errors
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
return {"detail": "Internal server error"}
2. Add Application Metrics
Integrate with Prometheus for metrics collection:
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
app = FastAPI()
# Setup Prometheus metrics
@app.on_event("startup")
async def startup():
Instrumentator().instrument(app).expose(app)
@app.get("/")
async def root():
return {"message": "Hello World"}
Error Handling & Validation
1. Implement Global Exception Handlers
Create custom exception handlers for a better user experience:
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import SQLAlchemyError
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
logger.warning(f"Validation error: {exc.errors()}")
return JSONResponse(
status_code=422,
content={"detail": "Validation Error", "errors": exc.errors()}
)
@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
logger.error(f"Database error: {str(exc)}", exc_info=True)
return JSONResponse(
status_code=500,
content={"detail": "Database error occurred. Please try again later."}
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
logger.warning(f"HTTP exception: {exc.detail} (status_code={exc.status_code})")
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
return JSONResponse(
status_code=500,
content={"detail": "Internal server error"}
)
2. Use Pydantic for Data Validation
Leverage Pydantic for input validation:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator, Field, EmailStr
from typing import Optional
import re
app = FastAPI()
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
password: str = Field(..., min_length=8)
age: Optional[int] = Field(None, gt=0, lt=120)
@validator('password')
def password_strength(cls, v):
if not re.search(r'[A-Z]', v):
raise ValueError('Password must contain at least one uppercase letter')
if not re.search(r'[a-z]', v):
raise ValueError('Password must contain at least one lowercase letter')
if not re.search(r'[0-9]', v):
raise ValueError('Password must contain at least one digit')
return v
@app.post("/users/")
async def create_user(user: UserCreate):
# Pydantic already validated the input
return {"username": user.username, "email": user.email}
Documentation and API Versioning
1. Configure API Documentation
Customize Swagger UI for better documentation:
from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html
app = FastAPI(
title="My Production API",
description="A fully production-ready FastAPI application",
version="1.0.0",
docs_url=None, # Disable the default docs URL
)
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url,
title=f"{app.title} - Swagger UI",
oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
swagger_js_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@4/swagger-ui-bundle.js",
swagger_css_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@4/swagger-ui.css",
)
2. Implement API Versioning
Use router prefixes for API versioning:
from fastapi import FastAPI, APIRouter
app = FastAPI()
# Version 1 router
v1_router = APIRouter(prefix="/api/v1")
@v1_router.get("/users")
async def get_users_v1():
return [{"name": "User 1", "email": "[email protected]"}]
# Version 2 router with new features
v2_router = APIRouter(prefix="/api/v2")
@v2_router.get("/users")
async def get_users_v2():
return [
{
"name": "User 1",
"email": "[email protected]",
"role": "admin", # New field in v2
"active": True # New field in v2
}
]
# Include both routers
app.include_router(v1_router)
app.include_router(v2_router)
Production Checklist Summary
Here's a quick summary of steps to prepare your FastAPI application for production:
-
Security
- ✅ Configure CORS properly
- ✅ Implement authentication and authorization
- ✅ Use environment variables for secrets
- ✅ Enable HTTPS
- ✅ Implement rate limiting
-
Performance
- ✅ Use async database connections
- ✅ Implement caching
- ✅ Set up connection pooling
-
Deployment
- ✅ Use Gunicorn with Uvicorn workers
- ✅ Containerize your application
- ✅ Set up health checks
-
Monitoring and Logging
- ✅ Configure comprehensive logging
- ✅ Add application metrics
-
Error Handling
- ✅ Implement global exception handlers
- ✅ Use Pydantic for data validation
-
Documentation
- ✅ Configure API documentation
- ✅ Implement API versioning
Additional Resources
- FastAPI Official Documentation
- Pydantic Documentation
- Uvicorn Documentation
- Gunicorn Documentation
- OWASP API Security Top 10
Exercises
-
Security Audit: Review one of your FastAPI applications and implement at least three security improvements from this checklist.
-
Performance Optimization: Profile your FastAPI application and identify performance bottlenecks. Implement caching and measure the improvements.
-
Monitoring Setup: Integrate Prometheus and Grafana with your FastAPI application to monitor request rates, response times, and error rates.
-
Containerization: Create a Docker Compose setup with your FastAPI application, a database, and Redis for caching.
-
Error Handling: Implement comprehensive error handling for your API, including global exception handlers and custom error responses.
By following this production checklist, you'll be well on your way to deploying a robust, secure, and high-performance FastAPI application.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)