FastAPI Production Deployment - What Actually Works

Your local setup works great until you deploy it and everything breaks.

I spent a weekend debugging 503 errors because I was using the wrong fucking worker class.

FastAPI Cloud: Still on the Waitlist

FastAPI Cloud exists

• it's built by the same team behind FastAPI
• but it's still in early access with a waiting list.

You can install fastapi[standard] and run fastapi deploy, but you need to get approved first.

What they promise:

• HTTPS without certificate hell
• Scales to zero (allegedly saves money)
• Readable logs
• Custom domains without AWS documentation torture
• Team access that doesn't involve IAM

Reality check: It's still invitation-only as of September 2025.

If you need to deploy today, you're doing containers like everyone else.

Container Deployment: Where Everyone Actually Ends Up

Since Fast

API Cloud is still gatekept, here's what actually works right now.

Reality: You're probably going to end up with containers.

Docker, Kubernetes, or some managed container service. It's not glamorous, but it's what pays the bills.

Here's the deployment flow that doesn't completely suck:

Development → Docker Build → Registry → Production
     ↓              ↓           ↓           ↓  
  Local Code → Container Image → DockerHub → ECS/K8s

Here's a Dockerfile that won't bite you in production (learned this the hard way):

FROM python:
3.12-slim

WORKDIR /app

## Install dependencies FIRST (Docker layer caching)
COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

## Copy application
COPY . .

## Non-root user (security scanners will yell otherwise)
RUN adduser --disabled-password --gecos '' appuser
USER appuser

## Gunicorn with Uvicorn workers (NOT just uvicorn)
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.

Uvicorn

Worker", "--bind", "0.0.0.0:8000"]

Critical shit that will save you:

• Multi-stage builds or your images will be 2GB
• Non-root user or your security team will kill you
• Gunicorn + Uvicorn workers or you'll crash under load
• Health checks or your orchestrator won't know you're dead
• Environment variables or you'll hardcode secrets like an idiot

Why Your Server Will Crash (And How To Fix It)

Uvicorn vs Gunicorn:

Why Your App Dies Under Load

Our API was getting 50 requests/second during lunch rush. Single Uvicorn worker crashed every damn day at 12:30pm until I switched to Gunicorn.

The difference? Single-threaded suicide vs actually handling concurrent requests.

Development (what you're probably using):

uvicorn main:app --reload --workers 1

• Single process, single thread
• Dies when one request blocks
• Fine for development, suicide in production

Production (what won't make you cry):

gunicorn main:app -w 4 -k uvicorn.workers.

UvicornWorker --bind 0.0.0.0:8000

Here's what's actually happening:

Load Balancer
    ↓
Gunicorn Master (watches workers, restarts dead ones)
    ├── Uvicorn Worker 1 (your app)
    ├── Uvicorn Worker 2 (your app)
    ├── Uvicorn Worker 3 (your app) 
    └── Uvicorn Worker 4 (your app)

Why this matters:

• One worker crashes?

Others keep serving requests

• Memory leak in one process? Doesn't kill everything
• Master process automatically restarts dead workers
• Can actually handle concurrent traffic

Worker count formula: Start with (2 × CPU cores) + 1.

I run 8 workers on 4-core machines and tune from there. Google Cloud Run is weird

• stick with single Uvicorn worker there since it manages processes differently.

Cloud Platform Reality Check

AWS: Pick Your Poison

Amazon ECS with Fargate: ECS Fargate works but costs more than you think.

I deployed a simple API and the bill was $120/month before I realized the load balancer alone costs $22/month. It scales automatically, which is nice when it works.

Amazon EKS: Full Kubernetes experience with all the YAML hell that entails.

I spent 3 weeks getting EKS working properly and the monthly bill made my CTO question everything. Only use if you actually need service mesh or your company is already committed to K8s.

AWS Lambda (Cold start roulette): Great for APIs that get hit 10 times a day.

Terrible for anything users expect to be fast. Cold starts are 500ms+ and will randomly piss off your users. I use mangum to adapt FastAPI for Lambda when I have to.

Azure and GCP:

The Alternatives

Azure Container Instances: Actually decent if you're already in the Microsoft ecosystem. Azure DevOps integration works well, but you'll pay extra for everything.

Google Cloud Run: My personal favorite for Fast

API.

Serverless containers that actually work, reasonable pricing, and scales from zero without the Lambda cold start penalty. Deploy with gcloud run deploy and you're done in 5 minutes.

Security: Don't Get Hacked

Authentication That Won't Get You Fired

I've seen FastAPI apps with hardcoded JWT secrets in production.

Don't be that person. Here's security that actually works:

from fastapi.security import HTTPBearer
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import Trusted

HostMiddleware
import jwt
import os

app = FastAPI()

## Middleware order matters 
- learned this debugging CORS issues for 6 hours
app.add_middleware(
    TrustedHostMiddleware, 
    allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # NOT ["*"] you absolute weapon
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["authorization", "content-type"],  # NOT ["*"]
)

security = HTTPBearer()

async def verify_token(credentials = Depends(security)):
    try:
        # Use RS256 for production, not HS256
        payload = jwt.decode(
            credentials.credentials, 
            PUBLIC_KEY,  # Not SECRET_KEY for RS256
            algorithms=["RS256"],
            options={"verify_exp":

 True, "verify_aud": True}
        )
        return payload
    except jwt.

ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.

InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")
    except Exception:
        # Log this shit so you know what's breaking
        logger.error(f"JWT verification failed: {str(e)}")
        raise HTTPException(status_code=401, detail="Authentication failed")

Secrets That Don't Leak

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    database_url: str
    jwt_public_key: str
    redis_url: str
    sentry_dsn: str | None = None
    log_level: str = "INFO"
    
    class Config:
        env_file = ".env"
        case_sensitive = False

@lru_cache()
def get_settings():
    return Settings()

Where to actually store secrets:

• AWS: Parameter Store (cheap) or Secrets Manager (expensive but auto-rotates)
• Azure: Key Vault (works well)
• GCP: Secret Manager (simple and cheap)
• Docker: Docker secrets or external injection

Monitoring:

Know When Shit's About to Hit the Fan

Error Tracking That Actually Helps

I learned about Sentry the hard way

• after spending 8 hours debugging a production issue that Sentry would have caught in 30 seconds.

Don't be me:

import sentry_sdk
from sentry_sdk.integrations.fastapi import Fast

ApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

sentry_sdk.init(
    dsn=os.getenv("SENTRY_DSN"),
    integrations=[
        FastApiIntegration(auto_enabling=True),
        SqlalchemyIntegration(),
    ],
    traces_sample_rate=0.1,  # Start low 
- 100% will kill performance
    environment="production",
    before_send=lambda event, hint: event if event.get('level') != 'info' else None
)

## Prometheus for metrics (because graphs are pretty)
from prometheus_fastapi_instrumentator import Instrumentator

instrumentator = Instrumentator(
    should_group_status_codes=False,  # You want to see 404s vs 500s
    should_ignore_untemplated=True,   # Ignore /favicon.ico spam
)
instrumentator.instrument(app).expose(app)

Metrics that saved my ass:

• Response time percentiles
• p95 tells you when users start complaining
• Error rates by endpoint
• /api/payments failing?

Priority 1

• Database connection pool usage
• hits 90%? You're about to crash
• Memory usage per worker
• grows without bounds? Memory leak
• Worker restart frequency
• restarting every 10 minutes? Something's wrong

Logging That Won't Make You Cry

Structured JSON logs are mandatory. I wasted 3 hours trying to debug an issue from unstructured logs that looked like someone vomited text:

import logging
import json
from datetime import datetime, timezone

class JSONFormatter(logging.

Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.get

Message(),
            "module": record.module,
            "function": record.func

Name,
            "line": record.lineno,  # You'll need this
        }
        # Add request context if available
        if hasattr(record, 'user_id'):
            log_entry['user_id'] = record.user_id
        if hasattr(record, 'request_id'):
            log_entry['request_id'] = record.request_id
        return json.dumps(log_entry)

## Production logging setup
logging.basic

Config(
    level=logging.

INFO,
    format="%(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
for handler in logger.handlers:
    handler.setFormatter(JSONFormatter())

## Add this to your FastAPI app
@app.middleware("http")
async def log_requests(request:

 Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() 
- start_time
    
    logger.info(
        "Request completed",
        extra={
            "method": request.method,
            "url": str(request.url),
            "status_code": response.status_code,
            "process_time": process_time
        }
    )
    return response

Database Connections:

Where Dreams Go to Die

Connection Pool Exhaustion: The 2am Wake-Up Call

Connection pool exhaustion is a bitch.

You'll see 'pool limit exceeded' right when your boss is demoing to investors. Set pool_size to 20, not the default 5.

from sqlalchemy import create_engine
from sqlalchemy.pool import Queue

Pool

## Don't use the defaults 
- they're too small
engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,  # Always available connections  
    max_overflow=30,  # Extra connections under load
    pool_pre_ping=True,  # Test connections before use
    pool_recycle=3600,  # Replace stale connections
    pool_timeout=30,  # Don't wait forever for a connection
)

Async Database That Won't Crash

from databases import Database
import asyncpg

## Async setup that handles production load
database = Database(
    DATABASE_URL,
    min_size=5,   # Always keep this many connections
    max_size=25,  # Scale up to this under load
    command_timeout=60,  # Kill long-running queries
    server_settings={
        "jit": "off",  # JIT causes unpredictable latency spikes
        "application_name": "fastapi_prod",  # Shows up in pg_stat_activity
    }
)

@app.on_event("startup")
async def startup():
    await database.connect()
    # Test the connection immediately
    await database.fetch_one("SELECT 1")

@app.on_event("shutdown") 
async def shutdown():
    await database.disconnect()

Critical gotcha: If you're using SQLAlchemy with async, make sure your connection string uses postgresql+asyncpg:// not postgresql://.

I spent 4 hours debugging "SSL SYSCALL error" messages before realizing I was using the sync driver. The error message was completely unhelpful.

CI/CD Pipeline Integration

GitHub Actions with Multiple Environments

name: Deploy FastAPI Application

on:
  push:
    branches: [main, staging]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:

- uses: actions/checkout@v4
      
- uses: actions/setup-python@v4
        with:
          python-version: '3.12'
      
- run: |
          pip install -r requirements.txt
          pytest tests/ -v
          
  deploy-staging:
    needs: test
    if: github.ref == 'refs/heads/staging'
    runs-on: ubuntu-latest
    steps:

- name:

 Deploy to staging
        run: |
          # Deploy to staging environment
          aws ecs update-service --cluster staging-cluster --service fastapi-service
          
  deploy-production:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    environment: production
    steps:

- name:

 Deploy to production
        run: |
          # Production deployment with blue-green strategy
          aws ecs update-service --cluster production-cluster --service fastapi-service

What Actually Works

Reality check: Most teams end up on ECS Fargate or Google Cloud Run.

Kubernetes is overkill unless you're Netflix. Lambda has cold start issues that will piss off users.

The basics that matter:

• Use Gunicorn with proper worker count (not single Uvicorn)
• Fix your database connection pool (20+ connections minimum)
• Set up monitoring that actually works (Sentry for errors)
• Don't hardcode secrets, use proper environment variables

Start simple. Deploy to Cloud Run or ECS. Add complexity only when you're forced to.

Fast

API Cloud will be great when it's publicly available. Until then, containers are what pays the bills.

The difference between a local FastAPI app and a production one is about 47 gotchas that will bite you at 3am. Here's the stuff that actually matters when your app needs to handle real traffic without falling over.

Advanced Server Configuration

Gunicorn Configuration That Won't Crash

Default Gunicorn settings are garbage for production. Here's a config that actually works:

## gunicorn_config.py - Production server configuration
import multiprocessing

## Worker configuration
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000  # Restart workers after handling this many requests
max_requests_jitter = 100  # Add randomization to prevent all workers restarting simultaneously

## Performance tuning
preload_app = True  # Load application before forking workers
keepalive = 5  # Keep connections alive for better performance

## Logging and monitoring
bind = "0.0.0.0:8000"
timeout = 30
graceful_timeout = 30
access_log_format = '%(h)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

Why these settings matter:

• Worker count: Start with (2 × CPU cores) + 1, then tune based on actual load testing (not some blog post formula)
• preload_app: Shares memory between workers, but will bite you if your app has global state (learned this the hard way)
• max_requests: Restarts workers to prevent memory leaks (Python's GC isn't perfect and will slowly eat your RAM)
• timeout settings: 30 seconds handles most requests, but if you have that one slow endpoint that takes 45 seconds, adjust accordingly

Memory Management and Why Your App Will Crash

Memory leaks in Python are real, and your FastAPI app will eat RAM until the OOM killer murders it. Here's how to prevent that:

import asyncio
import gc
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    
    # Configure event loop for better performance
    loop = asyncio.get_running_loop()
    loop.set_debug(False)  # Disable debug mode in production
    
    # Optimize garbage collection
    gc.set_threshold(700, 10, 10)  # Tune GC thresholds for your workload
    
    yield
    
    # Shutdown
    print("Shutting down...")
    # Close database connections, clear caches, etc.

app = FastAPI(lifespan=lifespan)

## Configure response compression
from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)

Database Performance and Why It Will Ruin Your Weekend

Connection Pool Hell

Database connections will fuck you over if you don't configure them properly. I've seen apps crash when connection pools hit their limit during traffic spikes:

## Advanced async database setup
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool, QueuePool

## Production database engine with optimized pooling
engine = create_async_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,  # Core connections always available (learned this after the 'pool limit exceeded' disaster)
    max_overflow=50,  # Additional connections under load
    pool_pre_ping=True,  # Validate connections before use
    pool_recycle=3600,  # Recycle connections every hour (connections go stale)
    connect_args={
        "server_settings": {
            "jit": "off",  # JIT causes unpredictable performance spikes
            "application_name": "fastapi_production",
        }
    }
)

## Learned this during a 3am AWS RDS maintenance window when our API started 
## throwing "connection already closed" errors. pool_pre_ping=True tests connections
## before using them so you don't get fucked by maintenance windows

## Session factory with proper configuration
async_session = async_sessionmaker(
    engine, 
    expire_on_commit=False,  # Keep objects accessible after commit
    autoflush=True,  # Automatically flush changes
)

## Dependency for database sessions
async def get_db():
    async with async_session() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Query Optimization and Caching

from functools import lru_cache
import redis.asyncio as redis
import json

## Redis connection for caching
redis_client = redis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)

## Cache decorator for expensive operations
def cache_result(expiration: int = 300):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            cache_key = f"cache:{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # Try to get from cache first
            cached_result = await redis_client.get(cache_key)
            if cached_result:
                return json.loads(cached_result)
            
            # Execute function and cache result
            result = await func(*args, **kwargs)
            await redis_client.setex(
                cache_key, 
                expiration, 
                json.dumps(result, default=str)
            )
            return result
        return wrapper
    return decorator

@cache_result(expiration=600)
async def get_popular_items(db: AsyncSession):
    # Expensive database query with proper indexing
    query = select(Item).where(Item.popularity_score > 0.8).options(
        joinedload(Item.category),  # Eager loading to prevent N+1 queries
        joinedload(Item.reviews)
    ).order_by(Item.popularity_score.desc()).limit(100)
    
    result = await db.execute(query)
    return [item.to_dict() for item in result.scalars().all()]

Advanced Security Implementation

Production Authentication System

from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
import secrets

## Secure password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

## JWT configuration with proper security
SECRET_KEY = secrets.token_urlsafe(32)  # Generate secure random key
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 30

class TokenManager:
    @staticmethod
    def create_access_token(data: dict) -> str:
        to_encode = data.copy()
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        to_encode.update({
            "exp": expire,
            "type": "access",
            "iat": datetime.utcnow(),
            "jti": secrets.token_urlsafe(16)  # Unique token ID for revocation
        })
        return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    
    @staticmethod
    def create_refresh_token(user_id: int) -> str:
        expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
        to_encode = {
            "user_id": user_id,
            "exp": expire,
            "type": "refresh",
            "iat": datetime.utcnow(),
            "jti": secrets.token_urlsafe(16)
        }
        return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

## Rate limiting 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.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/api/auth/login")
@limiter.limit("5/minute")  # Rate limit login attempts
async def login(request: Request, credentials: UserCredentials, db: AsyncSession = Depends(get_db)):
    # Secure login implementation with rate limiting and proper error handling
    pass

Security Middleware Stack

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time

## Security headers middleware
class SecurityHeadersMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            response = await self.app(scope, receive, send)
            # Add security headers
            response.headers["X-Content-Type-Options"] = "nosniff"
            response.headers["X-Frame-Options"] = "DENY"
            response.headers["X-XSS-Protection"] = "1; mode=block"
            response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
            return response
        return await self.app(scope, receive, send)

## Apply security middleware stack
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["yourdomain.com", "*.yourdomain.com", "localhost"]
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # Specific origins in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=86400  # Cache preflight requests for 24 hours
)

Monitoring and Observability

Comprehensive Application Metrics

from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
import time

## Custom metrics for business logic
REQUEST_COUNT = Counter('fastapi_requests_total', 'Total requests', ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('fastapi_request_duration_seconds', 'Request duration', ['method', 'endpoint'])
ACTIVE_CONNECTIONS = Gauge('fastapi_active_connections', 'Active connections')
DATABASE_CONNECTIONS = Gauge('fastapi_database_connections_active', 'Active database connections')

## Metrics middleware
class MetricsMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        
        start_time = time.time()
        method = scope["method"]
        path = scope["path"]
        
        try:
            ACTIVE_CONNECTIONS.inc()
            
            async def send_wrapper(message):
                if message["type"] == "http.response.start":
                    status_code = message["status"]
                    REQUEST_COUNT.labels(method=method, endpoint=path, status=status_code).inc()
                    REQUEST_DURATION.labels(method=method, endpoint=path).observe(time.time() - start_time)
                await send(message)
            
            await self.app(scope, receive, send_wrapper)
        finally:
            ACTIVE_CONNECTIONS.dec()

app.add_middleware(MetricsMiddleware)

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

Health Checks and Readiness Probes

from sqlalchemy import text

class HealthChecker:
    def __init__(self, db_engine, redis_client):
        self.db_engine = db_engine
        self.redis_client = redis_client
    
    async def check_database(self) -> bool:
        try:
            async with self.db_engine.begin() as conn:
                await conn.execute(text("SELECT 1"))
            return True
        except Exception:
            return False
    
    async def check_redis(self) -> bool:
        try:
            await self.redis_client.ping()
            return True
        except Exception:
            return False
    
    async def check_external_services(self) -> dict:
        # Check external API dependencies
        checks = {}
        # Add checks for external services your app depends on
        return checks

health_checker = HealthChecker(engine, redis_client)

@app.get("/health")
async def health_check():
    """Kubernetes liveness probe - basic app health"""
    return {"status": "healthy", "timestamp": datetime.utcnow()}

@app.get("/ready")  
async def readiness_check():
    """Kubernetes readiness probe - dependency health"""
    db_healthy = await health_checker.check_database()
    redis_healthy = await health_checker.check_redis()
    
    if not (db_healthy and redis_healthy):
        raise HTTPException(status_code=503, detail="Service dependencies not ready")
    
    return {
        "status": "ready",
        "checks": {
            "database": db_healthy,
            "redis": redis_healthy
        }
    }

Error Handling and Resilience

Production Error Management

from fastapi.exception_handlers import http_exception_handler
import logging
import traceback

## Structured logging for production
logger = logging.getLogger(__name__)

class ProductionExceptionHandler:
    @staticmethod
    async def http_exception_handler(request: Request, exc: HTTPException):
        # Log the exception with context
        logger.error(
            f"HTTP {exc.status_code}: {exc.detail}",
            extra={
                "url": str(request.url),
                "method": request.method,
                "headers": dict(request.headers),
                "user_agent": request.headers.get("user-agent")
            }
        )
        return await http_exception_handler(request, exc)
    
    @staticmethod
    async def general_exception_handler(request: Request, exc: Exception):
        # Log full traceback for unexpected errors
        logger.error(
            f"Unhandled exception: {str(exc)}",
            extra={
                "url": str(request.url),
                "method": request.method,
                "traceback": traceback.format_exc()
            }
        )
        
        # Return generic error to avoid leaking sensitive information
        return JSONResponse(
            status_code=500,
            content={"detail": "Internal server error"}
        )

## Register exception handlers
app.add_exception_handler(HTTPException, ProductionExceptionHandler.http_exception_handler)
app.add_exception_handler(Exception, ProductionExceptionHandler.general_exception_handler)

## Circuit breaker pattern for external service calls
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half_open
    
    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half_open"
            else:
                raise HTTPException(status_code=503, detail="Service temporarily unavailable")
        
        try:
            result = await func(*args, **kwargs)
            if self.state == "half_open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            
            raise e

This setup handles real traffic without crashing and tells you when shit's about to break. Implement these patterns gradually - don't try to do everything at once or you'll spend 3 weeks debugging configuration issues.

FastAPI apps need more than just the framework to survive production. You need proper connection pooling (learned this after pool exhaustion took down our API), security middleware that doesn't suck, monitoring that actually works, and error handling that doesn't leak sensitive info. Get these right and your app might actually scale without making you want to quit programming.