Fastapi Expert

FastAPI Expert

Production-ready FastAPI knowledge covering basic API development to planet-scale deployment.

Quick Start

Create a basic FastAPI application:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

# Run with: uvicorn main:app --reload

Core Topics

1. Database Operations

See: references/database.md

• SQLModel integration (recommended ORM)
• CRUD operations with dependency injection
• Async database operations
• Connection pooling
• Migrations with Alembic
• Neon Serverless PostgreSQL setup

2. Security & Authentication

See: references/security.md

• OAuth2 with password flow
• JWT token-based authentication
• Password hashing with Argon2
• OAuth2 scopes for permissions
• CORS configuration
• Rate limiting
• API key authentication
• Security best practices

3. Deployment & Scalability

See: references/deployment.md

• Docker containerization
• Kubernetes deployment
• Production server configuration (Uvicorn + Gunicorn)
• Horizontal pod autoscaling
• Performance monitoring with Prometheus
• Caching strategies with Redis
• Platform-specific guides (Vercel, AWS, GCP)

4. Advanced Features

See: references/advanced.md

• Dependency injection patterns
• Custom middleware
• WebSocket support
• Background tasks
• Request/response models with validation
• Streaming responses
• File uploads
• Testing strategies
• Event handlers

Project Templates

Use the provided production-ready templates in assets/:

FastAPI Project Structure

assets/project-template/
├── main.py           # Application entry point
├── config.py         # Settings management
├── database.py       # Database setup
├── models.py         # SQLModel models
└── auth.py           # Authentication logic

Copy the template to start a new project:

cp -r assets/project-template/* your-project/

Docker Deployment

Use assets/Dockerfile for containerizing your application with multi-stage builds and security best practices.

Kubernetes Deployment

Use assets/kubernetes-deployment.yaml for deploying to Kubernetes with:

• Deployment with 3 replicas
• Service with LoadBalancer
• Horizontal Pod Autoscaler
• Health and readiness probes

Common Patterns

Database CRUD with Session Dependency

from typing import Annotated
from fastapi import Depends
from sqlmodel import Session, select

SessionDep = Annotated[Session, Depends(get_session)]

@app.get("/users/{user_id}")
def get_user(user_id: int, session: SessionDep):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Protected Routes with Authentication

from typing import Annotated
from fastapi import Depends

CurrentUser = Annotated[User, Depends(get_current_active_user)]

@app.get("/users/me")
async def read_users_me(current_user: CurrentUser):
    return current_user

Background Tasks

from fastapi import BackgroundTasks

@app.post("/send-email/")
async def send_email(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_task, email)
    return {"message": "Email queued"}

Best Practices Checklist

Development

• [ ] Use type hints everywhere
• [ ] Implement request/response models with Pydantic
• [ ] Use dependency injection for shared logic
• [ ] Add proper error handling with HTTPException
• [ ] Use async/await for I/O operations

Security

• [ ] Hash passwords with Argon2
• [ ] Use JWT for authentication
• [ ] Implement OAuth2 scopes for authorization
• [ ] Configure CORS properly
• [ ] Store secrets in environment variables
• [ ] Enable HTTPS in production

Database

• [ ] Use SQLModel for ORM
• [ ] Implement connection pooling
• [ ] Use migrations (Alembic)
• [ ] Leverage dependency injection for sessions
• [ ] Add database indexes for performance

Deployment

• [ ] Multi-stage Dockerfile
• [ ] Non-root container user
• [ ] Health check endpoints
• [ ] Resource limits in Kubernetes
• [ ] Horizontal pod autoscaling
• [ ] Prometheus metrics
• [ ] Structured logging

Scalability Strategies

Async Operations

Always use async def for endpoints that perform I/O:

@app.get("/users/")
async def get_users():
    users = await fetch_from_db()
    return users

Caching

Implement Redis caching for frequently accessed data:

@cache(expire=600)
async def expensive_operation():
    # Heavy computation
    return result

Background Processing

Offload long-running tasks:

background_tasks.add_task(process_data, data)

Connection Pooling

Configure database connection pools:

engine = create_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30
)

Troubleshooting

Performance Issues

1. Enable Prometheus metrics to identify bottlenecks 2. Use async operations for all I/O 3. Implement caching with Redis 4. Optimize database queries (indexes, eager loading) 5. Enable GZip compression

Authentication Errors

1. Verify JWT secret key matches 2. Check token expiration time 3. Ensure password hashing is consistent 4. Validate CORS configuration

Database Connection Issues

1. Check connection string format 2. Verify connection pool settings 3. Test database reachability 4. Review firewall rules

Production Deployment Flow

1. Develop locally with auto-reload 2. Test with TestClient and pytest 3. Build Docker image 4. Push to container registry 5. Deploy to Kubernetes cluster 6. Monitor with Prometheus/Grafana 7. Scale with HPA based on metrics

Example: Complete CRUD API

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Annotated

# Database setup
engine = create_engine("sqlite:///database.db")

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

# Model
class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    description: str | None = None

# App
app = FastAPI()

@app.on_event("startup")
def on_startup():
    SQLModel.metadata.create_all(engine)

# CRUD endpoints
@app.post("/items/", response_model=Item)
def create_item(item: Item, session: SessionDep):
    session.add(item)
    session.commit()
    session.refresh(item)
    return item

@app.get("/items/", response_model=list[Item])
def read_items(session: SessionDep, skip: int = 0, limit: int = 100):
    items = session.exec(select(Item).offset(skip).limit(limit)).all()
    return items

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, session: SessionDep):
    item = session.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.patch("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item_update: Item, session: SessionDep):
    db_item = session.get(Item, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")

    item_data = item_update.model_dump(exclude_unset=True)
    for key, value in item_data.items():
        setattr(db_item, key, value)

    session.add(db_item)
    session.commit()
    session.refresh(db_item)
    return db_item

@app.delete("/items/{item_id}")
def delete_item(item_id: int, session: SessionDep):
    item = session.get(Item, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")

    session.delete(item)
    session.commit()
    return {"ok": True}

This skill provides everything needed to build production-ready FastAPI applications from basic CRUD to planet-scale deployments.

FastAPI Advanced Features

Dependency Injection System

Basic Dependencies

from fastapi import Depends

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

Class-Based Dependencies

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    return {"params": commons}

Dependency with Yield (Cleanup)

def get_db():
    db = Database()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def get_users(db: Database = Depends(get_db)):
    return db.get_all_users()

Sub-Dependencies

def query_extractor(q: str | None = None):
    return q

def query_or_default(query: str = Depends(query_extractor)):
    if query:
        return query
    return "default"

@app.get("/items/")
async def read_items(final_query: str = Depends(query_or_default)):
    return {"query": final_query}

Dependencies at Router Level

from fastapi import APIRouter

async def verify_token(x_token: str):
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="Invalid token")

router = APIRouter(dependencies=[Depends(verify_token)])

@router.get("/items/")
async def read_items():
    return [{"item": "Foo"}]

Global Dependencies

app = FastAPI(dependencies=[Depends(verify_api_key)])

Middleware

Custom HTTP Middleware

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Request Logging Middleware

import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"{request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Status: {response.status_code}")
    return response

CORS Middleware

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

GZip Compression

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

Trusted Host Middleware

from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

WebSocket Support

Basic WebSocket

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")

WebSocket with Authentication

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int, token: str):
    # Verify token
    user = verify_token(token)
    if not user:
        await websocket.close(code=1008)
        return

    await websocket.accept()

    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"User {client_id}: {data}")
    except WebSocketDisconnect:
        print(f"Client {client_id} disconnected")

WebSocket Connection Manager

from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left")

Background Tasks

Simple Background Task

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as log:
        log.write(f"{message}\n")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification sent"}

Multiple Background Tasks

@app.post("/process/")
async def process_data(data: Data, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, data.email)
    background_tasks.add_task(update_database, data.id)
    background_tasks.add_task(notify_webhooks, data)
    return {"status": "Processing started"}

Background Task with Dependencies

def process_order(order_id: int, session: Session = Depends(get_session)):
    order = session.get(Order, order_id)
    # Process order
    order.status = "completed"
    session.commit()

@app.post("/orders/")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    # Save order
    save_order(order)

    # Process in background
    background_tasks.add_task(process_order, order.id)
    return {"status": "Order created"}

Request and Response Models

Request Body Validation

from pydantic import BaseModel, Field, EmailStr

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: int = Field(..., ge=18, le=120)

@app.post("/users/")
async def create_user(user: UserCreate):
    return user

Response Model

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    # Don't include password in response

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # Save user with password
    saved_user = save_user(user)
    # Response automatically excludes password
    return saved_user

Multiple Response Models

from typing import Union

@app.get("/items/{item_id}", response_model=Union[Item, Error])
async def read_item(item_id: int):
    if item_id not in items_db:
        return Error(message="Item not found")
    return items_db[item_id]

Response Model with Exclusions

@app.get("/users/", response_model=List[User], response_model_exclude={"password"})
async def get_users():
    return users_db

Advanced Response Types

Streaming Response

from fastapi.responses import StreamingResponse
import io

def generate_csv():
    data = io.StringIO()
    data.write("id,name,email\n")
    for user in users:
        data.write(f"{user.id},{user.name},{user.email}\n")
    data.seek(0)
    return data

@app.get("/export/users")
async def export_users():
    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=users.csv"}
    )

File Response

from fastapi.responses import FileResponse

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = f"files/{filename}"
    return FileResponse(
        file_path,
        media_type="application/octet-stream",
        filename=filename
    )

Custom Response

from fastapi.responses import Response

@app.get("/custom")
async def custom_response():
    content = "<h1>Custom HTML</h1>"
    return Response(content=content, media_type="text/html")

Request Forms and Files

Form Data

from fastapi import Form

@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

File Upload

from fastapi import File, UploadFile

@app.post("/upload/")
async def upload_file(file: UploadFile = File()):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

Multiple File Upload

@app.post("/uploadfiles/")
async def upload_files(files: List[UploadFile] = File()):
    return {
        "filenames": [file.filename for file in files]
    }

Testing

Test Client

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

Testing with Database

import pytest
from sqlmodel import create_engine, Session

@pytest.fixture
def session():
    engine = create_engine("sqlite:///:memory:")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

def test_create_user(session):
    user = User(name="Test", email="test@example.com")
    session.add(user)
    session.commit()

    assert user.id is not None

Async Testing

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/items/")
    assert response.status_code == 200

Event Handlers

Startup Events

@app.on_event("startup")
async def startup_event():
    # Initialize database
    create_db_and_tables()
    # Warm up cache
    await warm_cache()
    print("Application started")

Shutdown Events

@app.on_event("shutdown")
async def shutdown_event():
    # Close database connections
    await database.disconnect()
    # Clean up resources
    print("Application shutting down")

Application Configuration

Settings Management

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My API"
    database_url: str
    secret_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

settings = Settings()

@app.get("/info")
async def info():
    return {"app_name": settings.app_name}

Dependency Injection for Settings

def get_settings():
    return Settings()

@app.get("/config")
async def get_config(settings: Settings = Depends(get_settings)):
    return {"app_name": settings.app_name}

FastAPI Database Operations

SQLModel Integration (Recommended)

SQLModel is the recommended ORM for FastAPI, combining SQLAlchemy and Pydantic.

Basic Setup

from sqlmodel import Field, Session, SQLModel, create_engine
from fastapi import Depends, FastAPI, HTTPException
from typing import Annotated

# Database engine
engine = create_engine("postgresql+psycopg://user:pass@localhost/db")

# Create tables
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

Model Definition

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    email: str = Field(unique=True)
    hashed_password: str
    disabled: bool = Field(default=False)

class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    description: str | None = None
    owner_id: int = Field(foreign_key="user.id")

Session Management with Dependency Injection

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

@app.get("/users/{user_id}")
def get_user(user_id: int, session: SessionDep):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

CRUD Operations

Create:

@app.post("/users/", response_model=User)
def create_user(user: User, session: SessionDep):
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

Read with Filtering:

from sqlmodel import select

@app.get("/users/", response_model=list[User])
def list_users(session: SessionDep, skip: int = 0, limit: int = 100):
    statement = select(User).offset(skip).limit(limit)
    users = session.exec(statement).all()
    return users

Update:

@app.patch("/users/{user_id}")
def update_user(user_id: int, user_update: UserUpdate, session: SessionDep):
    db_user = session.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")

    user_data = user_update.model_dump(exclude_unset=True)
    for key, value in user_data.items():
        setattr(db_user, key, value)

    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

Delete:

@app.delete("/users/{user_id}")
def delete_user(user_id: int, session: SessionDep):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    session.delete(user)
    session.commit()
    return {"ok": True}

Relationships

class Team(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    headquarters: str

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    team_id: int | None = Field(default=None, foreign_key="team.id")

# Query with joins
@app.get("/heroes/{hero_id}/team")
def get_hero_with_team(hero_id: int, session: SessionDep):
    statement = select(Hero, Team).where(Hero.id == hero_id).join(Team)
    result = session.exec(statement).first()
    if not result:
        raise HTTPException(status_code=404, detail="Hero not found")
    return result

Database Best Practices

1. Connection Pooling

from sqlmodel import create_engine

engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=20,          # Number of connections to maintain
    max_overflow=0,        # Max additional connections
    pool_timeout=30,       # Timeout for getting connection
    pool_recycle=3600,     # Recycle connections after 1 hour
)

2. Async Database Operations

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

async def get_session():
    async with AsyncSession(async_engine) as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.get(User, user_id)
    return result

3. Migrations with Alembic

# Install Alembic
pip install alembic

# Initialize
alembic init migrations

# Create migration
alembic revision --autogenerate -m "Add users table"

# Run migrations
alembic upgrade head

4. Database URL from Environment

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str

    class Config:
        env_file = ".env"

settings = Settings()
engine = create_engine(settings.database_url)

5. Transaction Management

@app.post("/transfer/")
def transfer_funds(transfer: Transfer, session: SessionDep):
    try:
        # Debit from source
        source = session.get(Account, transfer.source_id)
        source.balance -= transfer.amount

        # Credit to destination
        dest = session.get(Account, transfer.dest_id)
        dest.balance += transfer.amount

        session.commit()
        return {"status": "success"}
    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=400, detail=str(e))

Neon Serverless PostgreSQL

For serverless deployments, use Neon's connection pooling:

from sqlmodel import create_engine

# Use pooling connection string
engine = create_engine(
    "postgresql://user:pass@ep-xxx.us-east-2.aws.neon.tech/db?sslmode=require",
    connect_args={"sslmode": "require"}
)

Testing with In-Memory SQLite

from fastapi.testclient import TestClient

def test_create_user():
    # Use in-memory SQLite for tests
    test_engine = create_engine("sqlite:///:memory:")
    SQLModel.metadata.create_all(test_engine)

    def override_get_session():
        with Session(test_engine) as session:
            yield session

    app.dependency_overrides[get_session] = override_get_session

    client = TestClient(app)
    response = client.post("/users/", json={"name": "Test", "email": "test@example.com"})
    assert response.status_code == 200

FastAPI Deployment & Scalability

Docker Containerization

Basic Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Run with Uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Multi-Stage Build (Production)

# Build stage
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# Runtime stage
FROM python:3.11-slim

WORKDIR /app

# Copy dependencies from builder
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

# Copy application
COPY . .

# Create non-root user
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# Health check
HEALTHCHECK --interval=30s --timeout=3s \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose for Development

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/myapp
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    volumes:
      - ./:/app
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=myapp
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:

Kubernetes Deployment

Deployment Manifest

apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastapi
  template:
    metadata:
      labels:
        app: fastapi
    spec:
      containers:
      - name: fastapi
        image: myregistry/fastapi-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

Service Definition

apiVersion: v1
kind: Service
metadata:
  name: fastapi-service
spec:
  type: LoadBalancer
  selector:
    app: fastapi
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000

Horizontal Pod Autoscaler

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: fastapi-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: fastapi-app
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

ConfigMap for Configuration

apiVersion: v1
kind: ConfigMap
metadata:
  name: fastapi-config
data:
  app_name: "My FastAPI App"
  log_level: "INFO"
  cors_origins: "https://app.example.com,https://admin.example.com"

Secret for Sensitive Data

apiVersion: v1
kind: Secret
metadata:
  name: db-secret
type: Opaque
stringData:
  url: "postgresql://user:password@db-host:5432/mydb"
  secret_key: "your-jwt-secret-key"

Production Server Configuration

Uvicorn with Gunicorn (Multi-worker)

# Install
pip install gunicorn uvicorn[standard]

# Run with 4 worker processes
gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --keep-alive 5 \
  --log-level info \
  --access-logfile - \
  --error-logfile -

Calculate Optimal Workers

import multiprocessing

# Formula: (2 x CPU cores) + 1
workers = (2 * multiprocessing.cpu_count()) + 1

Uvicorn Standalone (Async)

uvicorn main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4 \
  --loop uvloop \
  --log-level info \
  --access-log \
  --use-colors

Scalability Strategies

1. Async Everything

# Use async for I/O operations
@app.get("/users/")
async def get_users():
    users = await fetch_users_from_db()  # Non-blocking
    return users

2. Connection Pooling

from sqlmodel import create_engine

engine = create_engine(
    DATABASE_URL,
    pool_size=20,           # Connections to maintain
    max_overflow=10,        # Additional connections allowed
    pool_timeout=30,        # Timeout for getting connection
    pool_recycle=3600,      # Recycle after 1 hour
    pool_pre_ping=True,     # Verify connections before use
)

3. Caching with Redis

from redis import asyncio as aioredis
from functools import wraps
import json

redis = aioredis.from_url("redis://localhost")

def cache(expire: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # Try cache first
            cached = await redis.get(cache_key)
            if cached:
                return json.loads(cached)

            # Execute function
            result = await func(*args, **kwargs)

            # Store in cache
            await redis.setex(cache_key, expire, json.dumps(result))
            return result
        return wrapper
    return decorator

@app.get("/expensive/")
@cache(expire=600)
async def expensive_operation():
    # Expensive computation
    return {"result": "data"}

4. Background Tasks for Async Processing

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Long-running task
    time.sleep(5)
    print(f"Email sent to {email}")

@app.post("/users/")
async def create_user(user: User, background_tasks: BackgroundTasks):
    # Save user (fast)
    save_user_to_db(user)

    # Send welcome email (slow, runs in background)
    background_tasks.add_task(send_email, user.email, "Welcome!")

    return {"status": "User created"}

5. Database Query Optimization

from sqlmodel import select
from sqlalchemy.orm import selectinload

# Eager loading to prevent N+1 queries
@app.get("/users-with-items/")
def get_users_with_items(session: SessionDep):
    statement = select(User).options(selectinload(User.items))
    users = session.exec(statement).all()
    return users

Performance Monitoring

Health Check Endpoint

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.get("/ready")
async def readiness_check(session: SessionDep):
    # Check database connection
    try:
        session.exec(select(User).limit(1))
        return {"status": "ready"}
    except:
        raise HTTPException(status_code=503, detail="Database not available")

Prometheus Metrics

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

# Add metrics endpoint
Instrumentator().instrument(app).expose(app)

# Metrics available at /metrics

Custom Middleware for Timing

import time
from fastapi import Request

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Deployment Checklist

Pre-deployment

• [ ] Set all secrets via environment variables
• [ ] Configure database migrations (Alembic)
• [ ] Set up logging (structured JSON logs)
• [ ] Configure CORS for production domains
• [ ] Enable HTTPS redirect
• [ ] Set up rate limiting
• [ ] Configure health check endpoints

Kubernetes Deployment

• [ ] Create Deployment with 3+ replicas
• [ ] Configure HPA for auto-scaling
• [ ] Set resource limits (CPU/memory)
• [ ] Add liveness and readiness probes
• [ ] Use ConfigMaps for configuration
• [ ] Use Secrets for sensitive data
• [ ] Set up Ingress for external access

Monitoring & Observability

• [ ] Enable Prometheus metrics
• [ ] Set up Grafana dashboards
• [ ] Configure log aggregation (ELK/Loki)
• [ ] Set up alerting (PagerDuty/Slack)
• [ ] Enable distributed tracing (Jaeger)

Security

• [ ] HTTPS only
• [ ] Strong JWT secret keys
• [ ] Rate limiting enabled
• [ ] CORS properly configured
• [ ] SQL injection prevention (use ORM)
• [ ] Input validation on all endpoints

Platform-Specific Guides

Vercel Deployment

# vercel.json
{
  "builds": [
    {
      "src": "main.py",
      "use": "@vercel/python"
    }
  ],
  "routes": [
    {
      "src": "/(.*)",
      "dest": "main.py"
    }
  ]
}

AWS ECS/Fargate

# task-definition.json
{
  "family": "fastapi-task",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "512",
  "memory": "1024",
  "containerDefinitions": [
    {
      "name": "fastapi",
      "image": "myregistry/fastapi:latest",
      "portMappings": [
        {
          "containerPort": 8000,
          "protocol": "tcp"
        }
      ],
      "environment": [
        {
          "name": "ENV",
          "value": "production"
        }
      ]
    }
  ]
}

Google Cloud Run

# Deploy to Cloud Run
gcloud run deploy fastapi-app \
  --image gcr.io/myproject/fastapi:latest \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated \
  --set-env-vars "DATABASE_URL=postgresql://..."

FastAPI Security Patterns

OAuth2 with Password Flow

Basic Setup

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# User models
class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    hashed_password: str

Password Hashing with Argon2

from argon2 import PasswordHasher

ph = PasswordHasher()

def hash_password(password: str) -> str:
    return ph.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    try:
        ph.verify(hashed_password, plain_password)
        return True
    except:
        return False

User Authentication

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$...",
        "disabled": False,
    }
}

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

JWT Token-Based Authentication

Token Creation

from datetime import datetime, timedelta
from jose import JWTError, jwt

SECRET_KEY = "your-secret-key-here-use-env-variable"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Token(BaseModel):
    access_token: str
    token_type: str

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)

    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

Token Verification

from typing import Annotated

async def get_current_user(token: Annotated[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
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

Login Endpoint

@app.post("/token", response_model=Token)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

Protected Routes

@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user

OAuth2 Scopes for Permissions

Define Scopes

from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "items:read": "Read items",
        "items:write": "Create and update items",
        "users:read": "Read user data",
        "users:write": "Modify user data",
    }
)

Scope-Based Authorization

from fastapi import Security

async def get_current_user(
    security_scopes: SecurityScopes,
    token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"

    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        token_scopes = payload.get("scopes", [])
    except JWTError:
        raise credentials_exception

    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )

    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items:read"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]

CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",
        "https://yourdomain.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["*"],
    expose_headers=["X-Total-Count"],
    max_age=600,  # Cache preflight for 10 minutes
)

Rate Limiting

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.get("/limited/")
@limiter.limit("5/minute")
async def limited_route(request: Request):
    return {"message": "Rate limited endpoint"}

API Key Authentication

from fastapi import Security
from fastapi.security import APIKeyHeader

API_KEY_HEADER = APIKeyHeader(name="X-API-Key")

def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if api_key != "your-secret-api-key":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API Key"
        )
    return api_key

@app.get("/protected/")
def protected_route(api_key: str = Depends(verify_api_key)):
    return {"message": "Access granted"}

Security Best Practices

1. Environment Variables for Secrets

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

2. HTTPS Only in Production

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

if not settings.debug:
    app.add_middleware(HTTPSRedirectMiddleware)

3. Secure Headers

from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
)

4. Input Validation

from pydantic import BaseModel, Field, EmailStr

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    email: EmailStr
    password: str = Field(..., min_length=8)

5. SQL Injection Prevention

SQLModel/SQLAlchemy automatically prevents SQL injection when using ORM methods. Avoid raw SQL:

# GOOD - Parameterized query
statement = select(User).where(User.username == username)

# BAD - String concatenation (vulnerable)
# query = f"SELECT * FROM users WHERE username = '{username}'"