← FastAPI

Deployment & Production

← Deployment & Production

Select a question from the sidebar

15 questions in this topic

Question 1 / 15

How do you configure Uvicorn for production?

Uvicorn is the recommended ASGI server for FastAPI. In development you run it with --reload, but production requires different settings for performance, security, and reliability.

Critical production flags:

Flag	Dev	Production
`--reload`	Yes	Never (file watchers waste CPU)
`--workers`	1	2-4 per CPU core
`--host`	127.0.0.1	0.0.0.0 (behind reverse proxy)
`--port`	8000	8000 (reverse proxy on 80/443)
`--log-level`	debug	info or warning
`--access-log`	Yes	Disable if reverse proxy logs
`--proxy-headers`	No	Yes (behind Nginx/ALB)
`--forwarded-allow-ips`	N/A	Reverse proxy IPs

Worker count formula:

A common guideline is (2 * CPU_cores) + 1. For I/O-bound apps (most APIs), this works well. For CPU-bound apps, keep workers equal to CPU cores.

Performance tuning:

--loop uvloop: Use uvloop for faster async I/O (Linux only)
--http httptools: Use httptools for faster HTTP parsing
--limit-concurrency: Limit max concurrent connections
--timeout-keep-alive: Default 5s, increase for long-polling

Security:

Always run behind a reverse proxy (Nginx, Caddy, or cloud load balancer)
Use --proxy-headers so Uvicorn trusts X-Forwarded-For headers
Set --forwarded-allow-ips to your reverse proxy's IP

Pseudo Code

# Development
uvicorn app:app --reload --log-level debug

# Production
uvicorn app:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 4 \
    --loop uvloop \
    --http httptools \
    --proxy-headers \
    --forwarded-allow-ips "172.17.0.1" \
    --log-level warning \
    --no-access-log

Code

# --- uvicorn_config.py: Programmatic configuration ---
import multiprocessing
import uvicorn

def get_worker_count() -> int:
    """Calculate optimal worker count based on CPU cores."""
    cores = multiprocessing.cpu_count()
    return (2 * cores) + 1

if __name__ == "__main__":
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        workers=get_worker_count(),
        loop="uvloop",
        http="httptools",
        proxy_headers=True,
        forwarded_allow_ips="*",  # Restrict in production!
        log_level="info",
        access_log=False,
        timeout_keep_alive=5,
        limit_concurrency=1000,
        limit_max_requests=10000,  # Restart worker after N requests
    )

# --- app/main.py ---
from fastapi import FastAPI

app = FastAPI(
    title="My API",
    docs_url=None,       # Disable Swagger in production
    redoc_url=None,      # Disable ReDoc in production
    openapi_url=None,    # Disable OpenAPI schema in production
)

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

# --- Command line production examples ---
# Basic production:
# uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# Optimized production (Linux):
# uvicorn app.main:app \
#     --host 0.0.0.0 --port 8000 \
#     --workers 9 \
#     --loop uvloop --http httptools \
#     --proxy-headers \
#     --forwarded-allow-ips "10.0.0.0/8" \
#     --log-level warning --no-access-log \
#     --limit-concurrency 1000 \
#     --limit-max-requests 10000

# With SSL (direct, no reverse proxy):
# uvicorn app.main:app \
#     --host 0.0.0.0 --port 443 \
#     --ssl-keyfile /path/to/key.pem \
#     --ssl-certfile /path/to/cert.pem

Key Points

•Never use --reload in production -- it wastes CPU on file watching
•Use --workers with (2 * CPU_cores) + 1 as a starting point
•Always enable --proxy-headers when behind a reverse proxy for correct client IPs
•Use uvloop and httptools on Linux for significant performance gains

Gotchas

•uvloop is not available on Windows -- Uvicorn falls back to the default asyncio loop
•--forwarded-allow-ips='*' trusts all proxies -- restrict to your actual proxy IPs in production
•Disabling docs_url/redoc_url/openapi_url in production prevents API schema exposure but also breaks interactive testing
•--limit-max-requests restarts workers after N requests to mitigate memory leaks, but causes brief request drops during restart

For production, run Uvicorn with --workers (2*CPU+1), --proxy-headers, --loop uvloop, --http httptools, and --no-access-log. Never use --reload. Always run behind a reverse proxy (Nginx/Caddy).

•Use --workers, --proxy-headers, uvloop, and httptools for production

# Production command
# uvicorn app.main:app \
#     --host 0.0.0.0 --port 8000 \
#     --workers 4 \
#     --loop uvloop --http httptools \
#     --proxy-headers \
#     --log-level warning --no-access-log

import uvicorn
if __name__ == "__main__":
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        workers=4,
        proxy_headers=True,
    )

Never use --reload in production -- it enables a file watcher that consumes CPU

Question 2 / 15

How do you use Gunicorn with Uvicorn workers?

Gunicorn is a mature process manager that can spawn and manage multiple Uvicorn worker processes. This combination gives you Gunicorn's robust process management (graceful restarts, worker recycling, signal handling) with Uvicorn's async ASGI performance.

Why Gunicorn + Uvicorn:

Gunicorn manages processes: Restarts crashed workers, handles signals (SIGHUP for reload, SIGTERM for shutdown)
Uvicorn handles async: Each worker runs an async event loop
Uvicorn's built-in --workers flag uses multiprocessing.Process which lacks Gunicorn's supervision features

Configuration:

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

-k uvicorn.workers.UvicornWorker: Use Uvicorn as the worker class
-w 4: Number of worker processes
-b 0.0.0.0:8000: Bind address

Gunicorn config file (`gunicorn.conf.py`):

Using a config file is cleaner than command-line flags and can include Python logic.

When to use Gunicorn:

Linux/macOS production deployments
When you need graceful worker recycling (max_requests)
When you need preload_app for shared memory
When you need signal-based reload (SIGHUP)

When NOT to use Gunicorn:

Windows (Gunicorn does not support Windows)
Docker containers with a single worker (use Uvicorn directly)
Development (use uvicorn --reload instead)

Pseudo Code

# Command line
gunicorn app.main:app \
    -k uvicorn.workers.UvicornWorker \
    -w 4 \
    -b 0.0.0.0:8000

# With config file
gunicorn app.main:app -c gunicorn.conf.py

Code

# --- gunicorn.conf.py ---
import multiprocessing

# Server socket
bind = "0.0.0.0:8000"

# Worker processes
workers = (2 * multiprocessing.cpu_count()) + 1
worker_class = "uvicorn.workers.UvicornWorker"

# Worker lifecycle
max_requests = 10000        # Restart worker after N requests (leak mitigation)
max_requests_jitter = 1000  # Random jitter to prevent all workers restarting at once
timeout = 120               # Kill worker if it does not respond in 120s
graceful_timeout = 30       # Wait 30s for worker to finish requests on restart
keepalive = 5               # Keep-alive timeout for connections

# Preload application (shared memory, faster fork)
preload_app = True

# Logging
accesslog = "-"          # Log to stdout
errorlog = "-"           # Log to stderr
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# Process naming
proc_name = "fastapi-app"

# Security
limit_request_line = 8190
limit_request_fields = 100
limit_request_field_size = 8190

# Hooks
def on_starting(server):
    """Called just before the master process is initialized."""
    pass

def post_fork(server, worker):
    """Called just after a worker has been forked."""
    server.log.info(f"Worker spawned (pid: {worker.pid})")

def pre_exec(server):
    """Called before a new master process is forked (on SIGHUP)."""
    server.log.info("Forked child, re-executing.")

def worker_exit(server, worker):
    """Called when a worker exits."""
    server.log.info(f"Worker exited (pid: {worker.pid})")


# --- app/main.py ---
from fastapi import FastAPI

app = FastAPI()

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

# Run:
# gunicorn app.main:app -c gunicorn.conf.py

# Graceful reload (zero downtime):
# kill -HUP $(cat /var/run/gunicorn.pid)

# Graceful shutdown:
# kill -TERM $(cat /var/run/gunicorn.pid)

Key Points

•Gunicorn manages worker processes with supervision, Uvicorn handles async ASGI within each worker
•Use -k uvicorn.workers.UvicornWorker to combine Gunicorn's process management with Uvicorn's async performance
•max_requests with jitter recycles workers to mitigate memory leaks without simultaneous restarts
•preload_app=True loads the app once in the master process, then forks -- saves memory via copy-on-write

Gotchas

•Gunicorn does not support Windows -- use Uvicorn directly with --workers on Windows
•preload_app=True means all workers share the same app object at fork time -- database connections created before fork will be shared and corrupted
•The timeout setting kills workers that do not respond -- set it higher than your slowest endpoint
•SIGHUP to Gunicorn triggers a graceful reload, but config file changes require a full restart

Use gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 to run FastAPI with Gunicorn managing worker processes. Gunicorn provides supervision, graceful restarts, and worker recycling. Use a gunicorn.conf.py file for clean configuration.

•Gunicorn supervises workers; Uvicorn handles async ASGI in each worker

# gunicorn.conf.py
import multiprocessing

bind = "0.0.0.0:8000"
workers = (2 * multiprocessing.cpu_count()) + 1
worker_class = "uvicorn.workers.UvicornWorker"
max_requests = 10000
max_requests_jitter = 1000
timeout = 120

# Run: gunicorn app.main:app -c gunicorn.conf.py

Gunicorn does not work on Windows -- use Uvicorn's built-in --workers flag instead

Question 3 / 15

How do you create a Dockerfile for FastAPI (using official python:3.x-slim image)?

A production Dockerfile for FastAPI should use the official python:3.x-slim image (not the deprecated tiangolo/uvicorn-gunicorn-fastapi image). Multi-stage builds reduce the final image size.

Why NOT tiangolo/uvicorn-gunicorn-fastapi:

It is deprecated and no longer maintained for new Python versions
It bundles Gunicorn config that may not match your needs
Official Python images are better maintained, more secure, and more flexible
You have full control over your process manager and configuration

Best practices:

Multi-stage build: Install dependencies in a builder stage, copy only what is needed to the final stage
Non-root user: Run the app as a non-root user for security
Pin Python version: Use python:3.12-slim, not python:3-slim
Layer caching: Copy requirements.txt before source code so dependency installation is cached
Health check: Add a HEALTHCHECK instruction
No dev dependencies: Only install production dependencies

Image size comparison:

Base Image	Size
python:3.12	~1GB
python:3.12-slim	~150MB
python:3.12-alpine	~50MB (but compatibility issues)

Pseudo Code

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
USER nonroot
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0"]

Code

# --- Dockerfile (production, multi-stage) ---

# Stage 1: Build dependencies
FROM python:3.12-slim AS builder

WORKDIR /app

# Install build dependencies (needed for some pip packages)
RUN apt-get update && \
    apt-get install -y --no-install-recommends gcc && \
    rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt


# Stage 2: Production image
FROM python:3.12-slim

WORKDIR /app

# Create non-root user
RUN groupadd -r appuser && \
    useradd -r -g appuser -d /app -s /sbin/nologin appuser

# Copy installed dependencies from builder
COPY --from=builder /install /usr/local

# Copy application code
COPY ./app ./app

# Set ownership
RUN chown -R appuser:appuser /app

# Switch to non-root user
USER appuser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

# Run with Gunicorn + Uvicorn workers
CMD ["gunicorn", "app.main:app", \
     "-k", "uvicorn.workers.UvicornWorker", \
     "-w", "4", \
     "-b", "0.0.0.0:8000", \
     "--access-logfile", "-", \
     "--error-logfile", "-"]


# --- requirements.txt ---
# fastapi==0.115.0
# uvicorn[standard]==0.30.0
# gunicorn==22.0.0


# --- .dockerignore ---
# __pycache__
# *.pyc
# .git
# .env
# .venv
# tests/
# docker-compose*.yml
# Dockerfile
# README.md


# Build and run:
# docker build -t fastapi-app .
# docker run -p 8000:8000 --name api fastapi-app

Key Points

•Use official python:3.12-slim, NOT the deprecated tiangolo/uvicorn-gunicorn-fastapi image
•Multi-stage builds reduce final image size by excluding build tools
•Always run as a non-root user for security
•Copy requirements.txt before source code to leverage Docker layer caching

Gotchas

•Alpine-based images (python:3.12-alpine) can have compatibility issues with C extensions -- prefer slim
•The HEALTHCHECK instruction only works in Docker -- Kubernetes uses its own liveness/readiness probes
•If using preload_app with Gunicorn, database connections created in the master process will be shared by forked workers -- initialize connections after fork
•Do not store secrets in the Docker image -- use environment variables or secret managers at runtime

Use python:3.12-slim as the base image (never the deprecated tiangolo/uvicorn-gunicorn-fastapi). Use multi-stage builds, a non-root user, and copy requirements.txt first for layer caching. Run with Gunicorn + Uvicorn workers.

•Use official python:3.x-slim -- the tiangolo image is deprecated

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN groupadd -r app && useradd -r -g app app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app ./app
USER app
EXPOSE 8000
CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker", "-w", "4", "-b", "0.0.0.0:8000"]

Alpine images can break C extensions -- prefer slim for reliable builds

Question 4 / 15

How do you manage environment variables and configuration?

FastAPI applications should follow the 12-Factor App methodology: store configuration in environment variables, never in code. Pydantic's BaseSettings (from pydantic-settings) is the recommended way to manage configuration.

Pydantic Settings features:

Automatic env var loading: Field names map to environment variable names (case-insensitive)
Type validation: Environment variables (always strings) are automatically cast to the declared type
Default values: Provide sensible defaults for development
.env file support: Load from .env files in development
Nested models: Group related settings (database, redis, etc.)
Secret file support: Read secrets from files (useful for Docker secrets)

Priority order (highest to lowest):

Init arguments passed directly
Environment variables
.env file values
Default values in the model

Best practices:

Use a single Settings class as the source of truth
Cache the settings instance with @lru_cache for performance
Never commit .env files to version control
Use separate .env files per environment (.env.dev, .env.prod)
Validate all settings at startup, not when first accessed

Pseudo Code

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    database_url: str
    redis_url: str = "redis://localhost"
    debug: bool = False

    model_config = SettingsConfigDict(env_file=".env")

@lru_cache
def get_settings():
    return Settings()

Code

# --- app/config.py ---
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, field_validator
from functools import lru_cache


class DatabaseSettings(BaseSettings):
    """Nested settings for database configuration."""
    url: str = "sqlite:///./app.db"
    pool_size: int = 5
    max_overflow: int = 10
    echo: bool = False

    model_config = SettingsConfigDict(env_prefix="DB_")


class Settings(BaseSettings):
    """Main application settings.

    Environment variables override defaults.
    Example: APP_NAME=MyAPI sets app_name.
    """
    # App
    app_name: str = "FastAPI App"
    debug: bool = False
    environment: str = "development"  # development, staging, production
    secret_key: str = "change-me-in-production"
    allowed_origins: list[str] = ["http://localhost:3000"]

    # Server
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 1

    # Database
    database_url: str = "sqlite:///./app.db"
    db_pool_size: int = 5

    # Redis
    redis_url: str = "redis://localhost:6379/0"

    # External services
    smtp_host: str = "localhost"
    smtp_port: int = 587
    api_key: str = ""

    @field_validator("environment")
    @classmethod
    def validate_environment(cls, v: str) -> str:
        allowed = {"development", "staging", "production"}
        if v not in allowed:
            raise ValueError(f"environment must be one of {allowed}")
        return v

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",  # Ignore unknown env vars
    )


@lru_cache
def get_settings() -> Settings:
    """Cached settings instance. Created once, reused everywhere."""
    return Settings()


# --- app/main.py ---
from fastapi import FastAPI, Depends
from typing import Annotated

# Validate settings at import time (fail fast)
settings = get_settings()

app = FastAPI(
    title=settings.app_name,
    debug=settings.debug,
)


@app.get("/config")
async def show_config(
    settings: Annotated[Settings, Depends(get_settings)],
):
    """Inject settings via dependency for testability."""
    return {
        "app_name": settings.app_name,
        "environment": settings.environment,
        "debug": settings.debug,
        # Never expose secrets!
    }


# --- .env (do NOT commit to git) ---
# APP_NAME=My Production API
# DEBUG=false
# ENVIRONMENT=production
# SECRET_KEY=super-secret-key-here
# DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/mydb
# REDIS_URL=redis://redis:6379/0
# ALLOWED_ORIGINS=["https://myapp.com","https://www.myapp.com"]


# --- Testing: override settings ---
# def get_test_settings():
#     return Settings(database_url="sqlite:///test.db", debug=True)
# app.dependency_overrides[get_settings] = get_test_settings

Key Points

•Use pydantic-settings BaseSettings to load and validate configuration from environment variables
•Cache the Settings instance with @lru_cache -- it is created once and reused
•Inject settings via Depends(get_settings) for testability -- override in tests with dependency_overrides
•Never commit .env files or secrets to version control

Gotchas

•pydantic-settings is a separate package -- install with pip install pydantic-settings
•@lru_cache means changes to environment variables after startup are not reflected -- restart the app to pick up changes
•list[str] fields in environment variables must be valid JSON: ALLOWED_ORIGINS=["http://localhost"]
•env_prefix applies to all fields in a nested model -- DB_URL, DB_POOL_SIZE, etc.

Use pydantic-settings BaseSettings to load configuration from environment variables with type validation. Cache with @lru_cache, inject via Depends(get_settings) for testability, and use .env files for development.

•BaseSettings auto-loads env vars with type validation; @lru_cache avoids re-reading

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    database_url: str = "sqlite:///app.db"
    debug: bool = False
    secret_key: str = "change-me"

    model_config = SettingsConfigDict(env_file=".env")

@lru_cache
def get_settings():
    return Settings()

pydantic-settings is a separate pip install -- it is not included with pydantic

Question 5 / 15

How do you implement structured logging for FastAPI?

Structured logging outputs log entries as JSON objects instead of unstructured text. This makes logs machine-parseable, searchable, and compatible with log aggregation tools (ELK, Datadog, CloudWatch).

Why structured logging:

Searchable: Query by field (request_id, user_id, status_code)
Parseable: Log aggregators can index JSON fields automatically
Consistent: Every log entry has the same format
Contextual: Attach request context (correlation ID, user, endpoint)

Popular libraries:

structlog: The most Pythonic structured logger, works great with FastAPI
python-json-logger: Adds JSON formatting to stdlib logging
loguru: Popular alternative with nice defaults

Key patterns:

Request ID middleware: Generate or extract a request ID for every request
Context binding: Attach request-specific data to all log entries
Correlation IDs: Trace a request across multiple services
Log levels: Use appropriate levels (DEBUG, INFO, WARNING, ERROR, CRITICAL)

Integration with Uvicorn:

Uvicorn has its own loggers (uvicorn.error, uvicorn.access). You can configure them to use the same structured format for consistency.

Pseudo Code

import structlog

logger = structlog.get_logger()

@app.middleware("http")
async def log_middleware(request, call_next):
    logger.info("request_started", path=request.url.path)
    response = await call_next(request)
    logger.info("request_completed", status=response.status_code)
    return response

Code

# --- app/logging_config.py ---
import structlog
import logging
import sys
import uuid
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import time


def setup_logging(json_output: bool = True, log_level: str = "INFO"):
    """Configure structured logging with structlog."""
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    if json_output:
        # JSON output for production
        renderer = structlog.processors.JSONRenderer()
    else:
        # Human-readable for development
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
    )

    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers = [handler]
    root_logger.setLevel(log_level)

    # Quiet noisy loggers
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)


# --- Request logging middleware ---
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Generate or extract request ID
        request_id = request.headers.get(
            "X-Request-ID", str(uuid.uuid4())
        )

        # Bind context for all logs in this request
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
            client_ip=request.client.host if request.client else "unknown",
        )

        logger = structlog.get_logger()
        start_time = time.perf_counter()

        logger.info("request_started")

        try:
            response = await call_next(request)
            duration_ms = (time.perf_counter() - start_time) * 1000

            logger.info(
                "request_completed",
                status_code=response.status_code,
                duration_ms=round(duration_ms, 2),
            )

            # Add request ID to response headers
            response.headers["X-Request-ID"] = request_id
            return response

        except Exception as e:
            duration_ms = (time.perf_counter() - start_time) * 1000
            logger.error(
                "request_failed",
                error=str(e),
                error_type=type(e).__name__,
                duration_ms=round(duration_ms, 2),
            )
            raise


# --- app/main.py ---
app = FastAPI()

# Initialize structured logging
setup_logging(json_output=True, log_level="INFO")

# Add middleware
app.add_middleware(RequestLoggingMiddleware)

logger = structlog.get_logger()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    logger.info("fetching_user", user_id=user_id)
    return {"user_id": user_id, "name": "Alice"}

# Example JSON log output:
# {"request_id": "abc-123", "method": "GET", "path": "/users/42",
#  "event": "request_completed", "status_code": 200, "duration_ms": 1.23,
#  "timestamp": "2025-01-15T10:30:00Z", "level": "info"}

Key Points

•Structured logging outputs JSON for machine-parseable, searchable log entries
•Use structlog with contextvars to bind request-specific data (request_id, path, method) to all logs
•Middleware generates correlation IDs and logs request duration and status code
•Pass the request ID in the X-Request-ID response header for end-to-end tracing

Gotchas

•structlog.contextvars requires Python 3.7+ and must be cleared between requests to prevent context leakage
•Uvicorn's access logger must be configured separately or silenced to avoid duplicate log entries
•JSON logging increases log volume -- configure log levels carefully in production
•Logging inside async generators (StreamingResponse) may not have the request context since the response is already streaming

Use structlog with contextvars to output JSON log entries with request-specific context (request_id, path, duration). Add a middleware that binds context, logs request start/end, and passes correlation IDs via X-Request-ID headers.

•structlog + contextvars produces JSON logs with per-request context

import structlog
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import uuid, time

logger = structlog.get_logger()

class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        structlog.contextvars.bind_contextvars(
            request_id=str(uuid.uuid4()),
            path=request.url.path,
        )
        start = time.perf_counter()
        response = await call_next(request)
        logger.info("request", status=response.status_code,
                     duration_ms=round((time.perf_counter()-start)*1000, 2))
        return response

app = FastAPI()
app.add_middleware(LogMiddleware)

Clear contextvars between requests to prevent context from leaking across requests

Question 6 / 15

How do you implement health check endpoints?

Health check endpoints let load balancers, orchestrators (Kubernetes), and monitoring tools determine if your application is running and ready to serve traffic.

Types of health checks:

Liveness check (/health/live): Is the process alive and not deadlocked?
- Should be fast and lightweight (no external dependencies)
- If it fails, the orchestrator restarts the container
- Should always return 200 unless the app is truly broken
Readiness check (/health/ready): Can the app serve traffic?
- Checks external dependencies (database, Redis, external APIs)
- If it fails, the load balancer stops sending traffic but does not restart
- Returns 503 if any critical dependency is down
Startup check (/health/startup): Has the app finished initializing?
- Used during slow startups (loading ML models, warming caches)
- Prevents liveness probes from killing the app during initialization

Kubernetes probe mapping:

livenessProbe:
  httpGet: { path: /health/live, port: 8000 }
readinessProbe:
  httpGet: { path: /health/ready, port: 8000 }
startupProbe:
  httpGet: { path: /health/startup, port: 8000 }

Best practices:

Liveness checks should NOT check external dependencies
Readiness checks should have reasonable timeouts
Include dependency status in the response body for debugging
Use appropriate HTTP status codes (200 OK, 503 Service Unavailable)

Pseudo Code

@app.get("/health/live")
async def liveness():
    return {"status": "alive"}

@app.get("/health/ready")
async def readiness():
    db_ok = await check_db()
    redis_ok = await check_redis()
    if not all([db_ok, redis_ok]):
        raise HTTPException(503)
    return {"status": "ready"}

Code

from fastapi import FastAPI, Response
from contextlib import asynccontextmanager
from datetime import datetime, timezone
import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

# Track startup state
app_ready = False


# --- Dependency health checkers ---
async def check_database() -> dict:
    """Check database connectivity."""
    try:
        # Replace with real DB check: await db.execute("SELECT 1")
        await asyncio.sleep(0.01)  # Simulated
        return {"database": {"status": "up", "latency_ms": 10}}
    except Exception as e:
        return {"database": {"status": "down", "error": str(e)}}

async def check_redis() -> dict:
    """Check Redis connectivity."""
    try:
        # Replace with real check: await redis.ping()
        await asyncio.sleep(0.01)  # Simulated
        return {"redis": {"status": "up", "latency_ms": 5}}
    except Exception as e:
        return {"redis": {"status": "down", "error": str(e)}}

async def check_external_api() -> dict:
    """Check external API availability."""
    try:
        await asyncio.sleep(0.01)
        return {"external_api": {"status": "up"}}
    except Exception as e:
        return {"external_api": {"status": "down", "error": str(e)}}


# --- Liveness: Is the process alive? ---
@app.get("/health/live")
async def liveness():
    """Lightweight check. No external dependencies.
    If this fails, the container should be restarted."""
    return {
        "status": "alive",
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


# --- Readiness: Can we serve traffic? ---
@app.get("/health/ready")
async def readiness(response: Response):
    """Checks all critical dependencies.
    If this fails, traffic should be routed elsewhere."""
    checks = await asyncio.gather(
        check_database(),
        check_redis(),
    )

    dependencies = {}
    for check in checks:
        dependencies.update(check)

    all_healthy = all(
        dep["status"] == "up"
        for dep in dependencies.values()
    )

    if not all_healthy:
        response.status_code = 503

    return {
        "status": "ready" if all_healthy else "not_ready",
        "dependencies": dependencies,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


# --- Startup: Has initialization completed? ---
@app.get("/health/startup")
async def startup_check(response: Response):
    """Returns 503 until the app has fully started."""
    if not app_ready:
        response.status_code = 503
        return {"status": "starting"}
    return {"status": "started"}


# --- Combined health endpoint ---
@app.get("/health")
async def combined_health(response: Response):
    """Single endpoint with all health information."""
    checks = await asyncio.gather(
        check_database(),
        check_redis(),
        check_external_api(),
    )
    dependencies = {}
    for check in checks:
        dependencies.update(check)

    all_healthy = all(
        dep["status"] == "up" for dep in dependencies.values()
    )

    if not all_healthy:
        response.status_code = 503

    return {
        "status": "healthy" if all_healthy else "degraded",
        "version": "1.2.3",
        "dependencies": dependencies,
    }


# --- Lifespan: mark app as ready after init ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    global app_ready
    # Initialization (load models, warm caches, etc.)
    logger.info("Initializing...")
    await asyncio.sleep(0.1)  # Simulate slow startup
    app_ready = True
    logger.info("App is ready")
    yield
    app_ready = False

# Pass lifespan to FastAPI constructor instead:
# app = FastAPI(lifespan=lifespan)

Key Points

•Liveness checks should be fast and check only the process itself, not external dependencies
•Readiness checks verify all critical dependencies (DB, Redis) and return 503 if any are down
•Run dependency checks concurrently with asyncio.gather() to minimize total check time
•Kubernetes uses separate probes for liveness, readiness, and startup

Gotchas

•Liveness probes that check databases can cause cascading restarts if the DB is temporarily slow
•Health checks bypass authentication -- ensure they do not expose sensitive information
•Set reasonable timeouts on health checks -- a hanging check can block the load balancer
•A global app_ready flag with multiple workers requires each worker to set it independently

Implement /health/live (process alive, no dependencies), /health/ready (all dependencies OK), and /health/startup (initialization complete). Return 200 for healthy and 503 for unhealthy. Run dependency checks concurrently with asyncio.gather().

•Liveness = process alive; Readiness = dependencies OK; use 503 for unhealthy

from fastapi import FastAPI, Response
import asyncio

app = FastAPI()

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

@app.get("/health/ready")
async def ready(response: Response):
    db_ok = await check_db()  # Returns True/False
    if not db_ok:
        response.status_code = 503
    return {"status": "ready" if db_ok else "not_ready"}

Never check external dependencies in liveness probes -- it causes cascading restarts

Question 7 / 15

How do you set up a reverse proxy (Nginx) for FastAPI?

A reverse proxy sits between clients and your FastAPI server, handling TLS termination, load balancing, static files, rate limiting, and request buffering. Nginx is the most common choice.

Why use a reverse proxy:

TLS termination: Nginx handles HTTPS; FastAPI receives plain HTTP
Static files: Serve CSS/JS/images directly from Nginx (much faster)
Load balancing: Distribute traffic across multiple FastAPI workers
Buffering: Buffer slow client uploads before sending to FastAPI
Rate limiting: Protect against abuse at the proxy level
Security: Hide internal server details, add security headers

Key Nginx directives for FastAPI:

proxy_pass: Forward requests to FastAPI
proxy_set_header: Forward client IP, host, and protocol
proxy_http_version 1.1: Required for WebSocket support
proxy_buffering off: Required for SSE/streaming

WebSocket proxying:

WebSocket connections require special Nginx configuration: Upgrade and Connection headers must be passed through.

Common gotcha:

Without proxy_set_header X-Forwarded-For, FastAPI sees Nginx's IP (127.0.0.1) instead of the real client IP. Enable --proxy-headers in Uvicorn.

Pseudo Code

# Nginx config structure
http {
    upstream fastapi {
        server 127.0.0.1:8000;
    }
    server {
        listen 443 ssl;
        location / {
            proxy_pass http://fastapi;
        }
        location /ws {
            proxy_pass http://fastapi;
            # WebSocket headers
        }
    }
}

Code

# --- /etc/nginx/sites-available/fastapi ---

upstream fastapi_backend {
    # Multiple servers for load balancing
    server 127.0.0.1:8000;
    # server 127.0.0.1:8001;  # Add more for horizontal scaling
    # keepalive 32;  # Connection pooling to backend
}

server {
    listen 80;
    server_name api.example.com;
    # Redirect HTTP to HTTPS
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    # --- TLS ---
    ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;

    # --- Security headers ---
    add_header X-Frame-Options DENY;
    add_header X-Content-Type-Options nosniff;
    add_header X-XSS-Protection "1; mode=block";
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains";

    # --- Logging ---
    access_log /var/log/nginx/fastapi_access.log;
    error_log /var/log/nginx/fastapi_error.log;

    # --- Rate limiting ---
    limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;

    # --- API endpoints ---
    location / {
        limit_req zone=api burst=20 nodelay;

        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Timeouts
        proxy_connect_timeout 10s;
        proxy_read_timeout 60s;
        proxy_send_timeout 60s;
    }

    # --- WebSocket endpoints ---
    location /ws {
        proxy_pass http://fastapi_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket timeout (close idle connections after 1 hour)
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
    }

    # --- Server-Sent Events ---
    location /events {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Disable buffering for SSE
        proxy_buffering off;
        proxy_cache off;
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding off;
    }

    # --- Static files (served directly by Nginx) ---
    location /static {
        alias /var/www/app/static;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    # --- Client max body size ---
    client_max_body_size 10M;
}

# Test config: nginx -t
# Reload: nginx -s reload

Key Points

•Nginx handles TLS, static files, rate limiting, and load balancing -- FastAPI only handles API logic
•proxy_set_header X-Forwarded-For passes the real client IP to FastAPI (requires --proxy-headers in Uvicorn)
•WebSocket proxying requires proxy_http_version 1.1 and Upgrade/Connection headers
•Disable proxy_buffering for SSE/streaming endpoints to prevent Nginx from buffering the entire response

Gotchas

•Without proxy_set_header Host, FastAPI cannot generate correct URLs in responses
•The default proxy_read_timeout is 60s -- WebSocket connections will be closed after 60s of inactivity unless increased
•client_max_body_size must be set in Nginx AND in FastAPI if you have file upload limits
•HTTP/2 between Nginx and upstream is not supported -- only client-facing connections use HTTP/2

Nginx acts as a reverse proxy: handles TLS, forwards requests to FastAPI with proxy_pass, passes real client IPs with X-Forwarded-For, and requires special config for WebSocket (proxy_http_version 1.1 + Upgrade headers) and SSE (proxy_buffering off).

•Forward X-Forwarded-For for real client IPs; special headers for WebSocket

# /etc/nginx/sites-available/fastapi
server {
    listen 443 ssl;
    server_name api.example.com;

    ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    location /ws {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Default proxy_read_timeout of 60s kills idle WebSocket connections -- increase for long-lived connections

Question 8 / 15

How do you scale FastAPI applications horizontally?

Horizontal scaling means adding more server instances to handle increased load, as opposed to vertical scaling (adding more CPU/RAM to one server). FastAPI scales horizontally well because it is naturally stateless.

Prerequisites for horizontal scaling:

Stateless application: No in-memory state shared between requests
- Sessions: Use external stores (Redis, database)
- File uploads: Use object storage (S3, MinIO)
- Caches: Use distributed caches (Redis, Memcached)
- WebSocket connections: Use Redis Pub/Sub for cross-instance messaging
External data stores: Database, Redis, and file storage are separate services
Load balancer: Distribute traffic across instances
- Nginx, HAProxy, AWS ALB, or Kubernetes Ingress

Scaling strategies:

Strategy	Description
Multiple workers	Uvicorn `--workers 4` or Gunicorn
Multiple containers	Docker replicas behind a load balancer
Kubernetes	HPA (Horizontal Pod Autoscaler) based on CPU/memory/custom metrics
Serverless	AWS Lambda with Mangum adapter

What breaks when you scale:

In-memory caches (use Redis)
File-based sessions (use Redis/DB)
WebSocket connection managers (use Redis Pub/Sub)
Scheduled tasks (use distributed locks)
Database connection exhaustion (use connection pooling with PgBouncer)

Pseudo Code

# Docker Compose scaling
# docker compose up --scale api=4

# Kubernetes HPA
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
spec:
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        averageUtilization: 70

Code

# --- Stateless FastAPI app ready for horizontal scaling ---
from fastapi import FastAPI, Depends, Request
from contextlib import asynccontextmanager
from typing import Annotated
import redis.asyncio as aioredis
import json
import os


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Each instance connects to the SAME external Redis
    app.state.redis = aioredis.from_url(
        os.getenv("REDIS_URL", "redis://redis:6379"),
    )
    yield
    await app.state.redis.close()


app = FastAPI(lifespan=lifespan)


# --- Session stored in Redis (not in-memory) ---
async def get_redis(request: Request) -> aioredis.Redis:
    return request.app.state.redis


@app.post("/sessions")
async def create_session(
    redis: Annotated[aioredis.Redis, Depends(get_redis)],
):
    import uuid
    session_id = str(uuid.uuid4())
    await redis.setex(
        f"session:{session_id}",
        3600,  # 1 hour TTL
        json.dumps({"user_id": 42, "role": "admin"}),
    )
    return {"session_id": session_id}


@app.get("/sessions/{session_id}")
async def get_session(
    session_id: str,
    redis: Annotated[aioredis.Redis, Depends(get_redis)],
):
    data = await redis.get(f"session:{session_id}")
    if not data:
        from fastapi import HTTPException
        raise HTTPException(404, "Session not found")
    return json.loads(data)


# --- Distributed rate limiting ---
@app.get("/api/data")
async def get_data(
    request: Request,
    redis: Annotated[aioredis.Redis, Depends(get_redis)],
):
    client_ip = request.client.host
    key = f"rate:{client_ip}"
    count = await redis.incr(key)
    if count == 1:
        await redis.expire(key, 60)  # 60-second window
    if count > 100:
        from fastapi import HTTPException
        raise HTTPException(429, "Rate limit exceeded")
    return {"data": "value", "requests_remaining": 100 - count}


# --- Docker Compose for horizontal scaling ---
# docker-compose.yml:
# services:
#   api:
#     build: .
#     deploy:
#       replicas: 4
#     environment:
#       - REDIS_URL=redis://redis:6379
#   redis:
#     image: redis:7-alpine
#   nginx:
#     image: nginx:alpine
#     ports: ["80:80"]
#     depends_on: [api]

# Scale dynamically:
# docker compose up --scale api=8

Key Points

•Stateless design is the prerequisite -- move all shared state to external stores (Redis, database, S3)
•Each instance connects to the same Redis/database -- no in-memory state between requests
•Use distributed rate limiting (Redis INCR) instead of in-process counters
•Docker Compose replicas or Kubernetes HPA handle scaling the number of instances

Gotchas

•In-memory caches, WebSocket managers, and scheduled tasks break when scaling -- move them to Redis
•Database connection pools are per-instance -- 4 instances with pool_size=20 means 80 connections total, which may exceed your DB limit
•Distributed locks (Redis SETNX) are needed to prevent duplicate scheduled task execution
•Sticky sessions (session affinity) solve some problems but prevent even load distribution

Make your app stateless: move sessions, caches, and shared state to Redis or a database. Then scale by adding instances behind a load balancer (Docker Compose replicas, Kubernetes HPA). Each instance connects to the same external data stores.

•Stateless + external stores = horizontally scalable

# Stateless: session in Redis, not in memory
import redis.asyncio as aioredis

@app.post("/sessions")
async def create_session(redis=Depends(get_redis)):
    session_id = str(uuid.uuid4())
    await redis.setex(f"session:{session_id}", 3600, data)
    return {"session_id": session_id}

# Scale: docker compose up --scale api=4

Database connection pools multiply with instances -- 4 instances x 20 pool = 80 connections

Question 9 / 15

How do you handle graceful shutdown?

Graceful shutdown means the server stops accepting new requests but finishes processing in-flight requests before exiting. This prevents dropped connections and data corruption.

How it works in Uvicorn/FastAPI:

Server receives SIGTERM (or SIGINT)
Uvicorn stops accepting new connections
In-flight requests are given time to complete (configurable timeout)
The lifespan shutdown code runs (after yield in the lifespan context manager)
Worker exits

Key settings:

Uvicorn: --timeout-graceful-shutdown 30 (default: None, waits forever)
Gunicorn: graceful_timeout = 30 (default: 30 seconds)
Kubernetes: terminationGracePeriodSeconds: 30 (default: 30 seconds)

What to clean up during shutdown:

Close database connection pools
Flush pending writes (logs, metrics)
Cancel background asyncio tasks
Close HTTP client sessions
Deregister from service discovery

Signal handling:

Uvicorn handles SIGTERM and SIGINT automatically. Do not override signal handlers -- use the lifespan context manager instead.

Pseudo Code

@asynccontextmanager
async def lifespan(app):
    # Startup
    db = await create_pool()
    bg_task = asyncio.create_task(periodic_job())
    yield
    # Shutdown (graceful)
    bg_task.cancel()
    await bg_task  # Wait for cancellation
    await db.close()
    await flush_metrics()

Code

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio
import logging
import signal

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ResourceManager:
    """Manages shared resources with proper cleanup."""

    def __init__(self):
        self.db_pool = None
        self.http_client = None
        self.background_tasks: list[asyncio.Task] = []

    async def startup(self):
        logger.info("Starting up resources...")
        # Simulated resource creation
        self.db_pool = "connected"
        logger.info("Database pool created")

        import httpx
        self.http_client = httpx.AsyncClient()
        logger.info("HTTP client created")

        # Start background task
        task = asyncio.create_task(self._periodic_cleanup())
        self.background_tasks.append(task)
        logger.info("Background tasks started")

    async def shutdown(self):
        logger.info("Shutting down gracefully...")

        # 1. Cancel background tasks
        for task in self.background_tasks:
            task.cancel()
        # Wait for all tasks to finish cancellation
        await asyncio.gather(
            *self.background_tasks,
            return_exceptions=True,
        )
        logger.info("Background tasks cancelled")

        # 2. Close HTTP client
        if self.http_client:
            await self.http_client.aclose()
            logger.info("HTTP client closed")

        # 3. Close database pool (last, since other cleanup may need it)
        if self.db_pool:
            # await self.db_pool.close()
            self.db_pool = None
            logger.info("Database pool closed")

        logger.info("Graceful shutdown complete")

    async def _periodic_cleanup(self):
        try:
            while True:
                logger.info("Running periodic cleanup")
                await asyncio.sleep(300)  # Every 5 minutes
        except asyncio.CancelledError:
            logger.info("Periodic cleanup cancelled gracefully")
            # Perform final cleanup
            raise  # Re-raise to complete cancellation


resources = ResourceManager()


@asynccontextmanager
async def lifespan(app: FastAPI):
    await resources.startup()
    yield
    # Everything after yield runs during graceful shutdown
    await resources.shutdown()


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health():
    return {"status": "ok", "db": resources.db_pool is not None}


@app.get("/slow")
async def slow_endpoint():
    """Simulates a slow request -- tests graceful shutdown behavior."""
    logger.info("Slow request started")
    await asyncio.sleep(10)
    logger.info("Slow request completed")
    return {"status": "done"}


# --- Running with graceful shutdown ---
# Uvicorn:
# uvicorn app:app --timeout-graceful-shutdown 30

# Gunicorn:
# gunicorn app:app -k uvicorn.workers.UvicornWorker --graceful-timeout 30

# Docker (sends SIGTERM, then SIGKILL after stop_grace_period):
# docker stop --time 30 container_name

# Kubernetes:
# terminationGracePeriodSeconds: 30
# lifecycle:
#   preStop:
#     exec:
#       command: ["/bin/sh", "-c", "sleep 5"]  # Wait for LB to drain

Key Points

•Graceful shutdown finishes in-flight requests before exiting -- use the lifespan context manager for cleanup
•Cancel asyncio tasks, close HTTP clients, and close database pools during shutdown (in that order)
•Configure --timeout-graceful-shutdown in Uvicorn or graceful_timeout in Gunicorn
•Background tasks must handle CancelledError to perform their own cleanup

Gotchas

•Kubernetes sends SIGTERM then SIGKILL after terminationGracePeriodSeconds -- make sure your cleanup fits within this window
•A preStop hook with a small sleep (5s) gives the load balancer time to stop routing traffic before the app starts shutting down
•asyncio.gather with return_exceptions=True prevents one failed task from blocking the others
•Uvicorn's default graceful timeout is None (wait forever) -- always set an explicit timeout in production

Graceful shutdown finishes in-flight requests, then runs the code after yield in lifespan. Cancel background tasks, close clients, and close DB pools. Configure --timeout-graceful-shutdown in Uvicorn. In Kubernetes, add a preStop sleep to let the load balancer drain.

•Lifespan code after yield runs during shutdown -- cancel tasks and close connections

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(background_job())
    yield
    # Shutdown
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)
    # await db.close()
    # await http_client.aclose()

app = FastAPI(lifespan=lifespan)

# uvicorn app:app --timeout-graceful-shutdown 30

Set an explicit graceful timeout -- the default is unlimited, which can delay deployments

Question 10 / 15

How do you implement CI/CD for FastAPI applications?

CI/CD automates testing, building, and deploying your FastAPI application. A typical pipeline includes linting, testing, building a Docker image, and deploying to production.

Pipeline stages:

Lint: Check code style (ruff, mypy)
Test: Run pytest with coverage
Build: Build Docker image
Push: Push image to a container registry
Deploy: Update the production service

CI best practices for FastAPI:

Run tests with a real database (SQLite for unit tests, PostgreSQL for integration)
Use service containers for Redis, PostgreSQL in CI
Cache pip dependencies between runs
Run security scanning (Snyk, Trivy) on Docker images
Use matrix testing for multiple Python versions

Deployment strategies:

Rolling update: Replace instances one at a time (Kubernetes default)
Blue/green: Run old and new versions simultaneously, switch traffic
Canary: Send a small percentage of traffic to the new version first

GitHub Actions is the most common CI/CD for open-source Python projects.

Pseudo Code

# Pipeline stages:
# 1. Lint (ruff, mypy)
# 2. Test (pytest --cov)
# 3. Build Docker image
# 4. Push to registry
# 5. Deploy (kubectl, docker compose)

Code

# --- .github/workflows/ci.yml ---
name: CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: pip install ruff mypy
      - name: Lint with ruff
        run: ruff check .
      - name: Type check
        run: mypy app/ --ignore-missing-imports

  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: testdb
        ports: ["5432:5432"]
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      redis:
        image: redis:7-alpine
        ports: ["6379:6379"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Cache pip
        uses: actions/cache@v4
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('requirements*.txt') }}
      - name: Install dependencies
        run: pip install -r requirements.txt -r requirements-test.txt
      - name: Run tests
        env:
          DATABASE_URL: postgresql://postgres:test@localhost:5432/testdb
          REDIS_URL: redis://localhost:6379/0
        run: pytest --cov=app --cov-report=xml -v
      - name: Upload coverage
        uses: codecov/codecov-action@v4
        with:
          file: coverage.xml

  build-and-push:
    needs: [lint, test]
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Log in to Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}
      - name: Build and push
        uses: docker/build-push-action@v5
        with:
          push: true
          tags: |
            myuser/fastapi-app:latest
            myuser/fastapi-app:${{ github.sha }}

  deploy:
    needs: [build-and-push]
    runs-on: ubuntu-latest
    steps:
      - name: Deploy to production
        run: |
          # Option 1: SSH and docker compose pull
          # ssh deploy@server 'cd /app && docker compose pull && docker compose up -d'

          # Option 2: Kubernetes
          # kubectl set image deployment/api api=myuser/fastapi-app:${{ github.sha }}

          echo "Deploy step - configure for your infrastructure"

Key Points

•Run lint (ruff), type check (mypy), and tests (pytest --cov) in separate parallel jobs
•Use GitHub Actions service containers for PostgreSQL and Redis in integration tests
•Cache pip dependencies to speed up CI runs
•Tag Docker images with both latest and the commit SHA for traceability

Gotchas

•Service containers in GitHub Actions only work on ubuntu runners, not on macos or windows
•Secrets (DOCKER_PASSWORD, deploy keys) must be configured in GitHub repository settings, not in the workflow file
•The deploy step needs appropriate credentials and network access to your production environment
•Running all tests on every PR can be slow -- use path filters to skip CI for docs-only changes

Use GitHub Actions with separate jobs for lint (ruff), test (pytest with service containers for PostgreSQL/Redis), build (Docker), and deploy. Cache pip dependencies, tag images with commit SHA, and only deploy from the main branch.

•Separate jobs for lint, test, build, deploy with service containers for dependencies

# .github/workflows/ci.yml
name: CI
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env: { POSTGRES_PASSWORD: test }
        ports: ["5432:5432"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements.txt && pytest --cov=app -v

Service containers only work on ubuntu runners in GitHub Actions

Question 11 / 15

How do you monitor FastAPI applications in production?

Production monitoring for FastAPI covers three pillars of observability: metrics, logs, and traces.

1. Metrics (Prometheus):

Numeric measurements over time -- request rate, latency, error rate, resource usage.

Use prometheus-fastapi-instrumentator for automatic metric collection
Expose a /metrics endpoint for Prometheus to scrape
Key metrics: request count, request duration, in-flight requests, error rate

2. Logs (structured):

Event records with context -- use structlog or python-json-logger.

Already covered in the structured logging question

3. Traces (OpenTelemetry):

End-to-end request flow across services.

Use opentelemetry-instrumentation-fastapi for automatic tracing
Export to Jaeger, Zipkin, or cloud providers

The RED method (recommended for APIs):

Rate: Requests per second
Errors: Error rate (4xx, 5xx)
Duration: Request latency (p50, p95, p99)

Common alerting rules:

Error rate > 5% for 5 minutes
p99 latency > 2 seconds
Health check returning 503
High CPU/memory usage

Pseudo Code

# Prometheus metrics
from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app, "/metrics")

# Custom metric
from prometheus_client import Counter
orders = Counter("orders_total", "Total orders")

@app.post("/orders")
async def create_order():
    orders.inc()
    return {"ok": True}

Code

# --- Prometheus metrics ---
from fastapi import FastAPI, Request, Response
from prometheus_client import (
    Counter, Histogram, Gauge,
    generate_latest, CONTENT_TYPE_LATEST,
)
import time

app = FastAPI()

# Define metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
)
REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)
IN_FLIGHT = Gauge(
    "http_requests_in_flight",
    "Currently processing requests",
)


# --- Metrics middleware ---
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    if request.url.path == "/metrics":
        return await call_next(request)

    method = request.method
    endpoint = request.url.path

    IN_FLIGHT.inc()
    start = time.perf_counter()

    try:
        response = await call_next(request)
        REQUEST_COUNT.labels(
            method=method,
            endpoint=endpoint,
            status=response.status_code,
        ).inc()
        return response
    except Exception as e:
        REQUEST_COUNT.labels(
            method=method,
            endpoint=endpoint,
            status=500,
        ).inc()
        raise
    finally:
        duration = time.perf_counter() - start
        REQUEST_DURATION.labels(
            method=method,
            endpoint=endpoint,
        ).observe(duration)
        IN_FLIGHT.dec()


# --- Metrics endpoint for Prometheus to scrape ---
@app.get("/metrics")
async def metrics():
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST,
    )


# --- Custom business metrics ---
ORDERS_CREATED = Counter(
    "orders_created_total",
    "Total orders created",
    ["product_type"],
)

@app.post("/orders")
async def create_order():
    ORDERS_CREATED.labels(product_type="digital").inc()
    return {"status": "created"}


# --- OpenTelemetry tracing (optional) ---
# pip install opentelemetry-instrumentation-fastapi opentelemetry-exporter-otlp
#
# from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
# from opentelemetry import trace
# from opentelemetry.sdk.trace import TracerProvider
# from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# from opentelemetry.sdk.trace.export import BatchSpanProcessor
#
# provider = TracerProvider()
# provider.add_span_processor(
#     BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317"))
# )
# trace.set_tracer_provider(provider)
# FastAPIInstrumentor.instrument_app(app)


# --- Prometheus scrape config (prometheus.yml) ---
# scrape_configs:
#   - job_name: fastapi
#     scrape_interval: 15s
#     static_configs:
#       - targets: ["fastapi:8000"]


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

Key Points

•Use Counter for request counts, Histogram for latency distribution, Gauge for in-flight requests
•Expose a /metrics endpoint for Prometheus to scrape at regular intervals
•The RED method (Rate, Errors, Duration) covers the most important API metrics
•OpenTelemetry provides distributed tracing across multiple services

Gotchas

•High-cardinality labels (e.g., user_id, request_id) in Prometheus metrics cause memory explosion
•Histogram buckets must be chosen carefully -- the default may not match your API's latency profile
•The /metrics endpoint should not require authentication, but should be restricted to internal networks
•Middleware-based metrics do not capture errors in middleware itself -- add error handling around the middleware

Use Prometheus metrics (Counter, Histogram, Gauge) with a /metrics endpoint and a middleware that tracks request count, duration, and in-flight requests. Add OpenTelemetry for distributed tracing. Follow the RED method: Rate, Errors, Duration.

•Counter for totals, Histogram for latency, /metrics endpoint for Prometheus

from prometheus_client import Counter, Histogram, generate_latest
from fastapi import FastAPI, Response

app = FastAPI()

REQUESTS = Counter("http_requests_total", "Total requests", ["method", "status"])
DURATION = Histogram("http_request_duration_seconds", "Request duration")

@app.middleware("http")
async def track(request, call_next):
    import time
    start = time.perf_counter()
    response = await call_next(request)
    DURATION.observe(time.perf_counter() - start)
    REQUESTS.labels(request.method, response.status_code).inc()
    return response

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

Avoid high-cardinality labels (user_id, URL with IDs) -- they explode Prometheus memory

Question 12 / 15

How do you handle database migrations in production?

Alembic is the standard migration tool for SQLAlchemy-based FastAPI applications. It tracks schema changes as versioned migration scripts that can be applied and rolled back.

Setup:

pip install alembic
alembic init migrations -- creates the migration directory
Configure alembic.ini and env.py with your database URL
alembic revision --autogenerate -m "description" -- generate a migration
alembic upgrade head -- apply all migrations

Production workflow:

Developer creates a migration locally
Migration is code-reviewed and merged
CI runs migrations against a test database
Deployment runs alembic upgrade head before starting the app

Running migrations in Docker:

Use an entrypoint script or init container that runs migrations before the app starts.

Key principles:

Never edit models without a migration -- the database schema must match the code
Never autogenerate blindly -- review every generated migration
Test rollbacks -- every upgrade() should have a working downgrade()
Avoid destructive migrations in production -- rename, then deprecate, then drop

Safe migration patterns:

Add column with default: Safe (no lock)
Add index concurrently: Safe (use op.create_index(..., postgresql_concurrently=True))
Drop column: Dangerous (remove code references first, deploy, then drop)
Rename column: Dangerous (add new, copy data, drop old)

Pseudo Code

# Generate migration
alembic revision --autogenerate -m "add users table"

# Apply migrations
alembic upgrade head

# Rollback
alembic downgrade -1

# Docker entrypoint
alembic upgrade head && uvicorn app:app

Code

# --- migrations/env.py (key parts) ---
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.models import Base  # Import your SQLAlchemy models
from app.config import get_settings

settings = get_settings()
config = context.config

# Use the app's database URL
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata


def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,  # Detect column type changes
        )
        with context.begin_transaction():
            context.run_migrations()


# --- Example migration file ---
# migrations/versions/001_add_users.py
"""Add users table

Revision ID: 001abc
Create Date: 2025-01-15
"""
from alembic import op
import sqlalchemy as sa

revision = "001abc"
down_revision = None


def upgrade():
    op.create_table(
        "users",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("email", sa.String(255), nullable=False, unique=True),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.func.now(),
        ),
    )
    op.create_index("ix_users_email", "users", ["email"])


def downgrade():
    op.drop_index("ix_users_email", "users")
    op.drop_table("users")


# --- Safe column addition (no lock on large tables) ---
# migrations/versions/002_add_phone.py
def upgrade():
    op.add_column(
        "users",
        sa.Column("phone", sa.String(20), nullable=True),  # nullable=True is safe
    )

def downgrade():
    op.drop_column("users", "phone")


# --- Docker entrypoint script (entrypoint.sh) ---
# #!/bin/bash
# set -e
#
# echo "Running database migrations..."
# alembic upgrade head
#
# echo "Starting application..."
# exec gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000


# --- Dockerfile CMD ---
# COPY entrypoint.sh /entrypoint.sh
# RUN chmod +x /entrypoint.sh
# ENTRYPOINT ["/entrypoint.sh"]


# --- Testing migrations in CI ---
# pytest test that verifies migrations work:
import subprocess

def test_migrations_up_and_down():
    """Verify all migrations can be applied and rolled back."""
    # Apply all migrations
    result = subprocess.run(
        ["alembic", "upgrade", "head"],
        capture_output=True, text=True,
    )
    assert result.returncode == 0, result.stderr

    # Roll back all migrations
    result = subprocess.run(
        ["alembic", "downgrade", "base"],
        capture_output=True, text=True,
    )
    assert result.returncode == 0, result.stderr

Key Points

•Alembic tracks schema changes as versioned migration scripts with upgrade() and downgrade()
•Run alembic upgrade head before starting the app -- use a Docker entrypoint script
•Always review autogenerated migrations -- they can miss renames and produce destructive drops
•Add columns as nullable=True first, then backfill, then set NOT NULL in a separate migration

Gotchas

•Autogenerate does not detect renamed columns -- it generates a drop + add, which loses data
•Running migrations during deployment with multiple replicas can cause race conditions -- use a database advisory lock or run migrations in a separate init container
•compare_type=True is needed to detect column type changes, but can produce false positives
•Rolling back migrations in production is risky -- if upgrade() added data, downgrade() may not cleanly undo it

Use Alembic for database migrations. Generate with alembic revision --autogenerate, apply with alembic upgrade head. Run migrations in a Docker entrypoint before starting the app. Always review autogenerated scripts and test rollbacks.

•Alembic manages versioned migrations; run upgrade head before starting the app

# Generate migration
# alembic revision --autogenerate -m "add users table"

# Apply
# alembic upgrade head

# Docker entrypoint.sh
#!/bin/bash
set -e
alembic upgrade head
exec gunicorn app.main:app -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

Autogenerate detects renames as drop+add -- always review generated migrations to avoid data loss

Question 13 / 15

How do you implement HTTPS/TLS for FastAPI?

HTTPS encrypts traffic between the client and server. In production, TLS is typically terminated at the reverse proxy (Nginx, Caddy) or load balancer, not in the application itself.

Three approaches:

Reverse proxy termination (recommended): Nginx or Caddy handles TLS. FastAPI receives plain HTTP on an internal network. This is the most common and performant approach.
Uvicorn direct TLS: Uvicorn can serve HTTPS directly using --ssl-keyfile and --ssl-certfile. Useful for simple deployments without a reverse proxy.
Cloud load balancer: AWS ALB, GCP Load Balancer, or Cloudflare terminate TLS. Your app never sees HTTPS.

Let's Encrypt (free certificates):

Use certbot to obtain and auto-renew certificates
Caddy has built-in automatic HTTPS with Let's Encrypt
Nginx requires certbot plugin or manual setup

HTTPS redirect:

Force all HTTP traffic to HTTPS using Starlette's HTTPSRedirectMiddleware or Nginx's return 301.

HSTS (HTTP Strict Transport Security):

Tells browsers to always use HTTPS for your domain. Set the Strict-Transport-Security header.

Certificate types:

Let's Encrypt: Free, auto-renewable, widely trusted
Self-signed: For development/testing only
Commercial: For extended validation (EV) certificates

Pseudo Code

# Option 1: Nginx terminates TLS
# Nginx: listen 443 ssl;
# FastAPI: listen 8000 (plain HTTP)

# Option 2: Uvicorn direct TLS
uvicorn app:app --ssl-keyfile key.pem --ssl-certfile cert.pem

# Option 3: HTTPS redirect middleware
app.add_middleware(HTTPSRedirectMiddleware)

Code

# --- Option 1: HTTPS redirect middleware ---
from fastapi import FastAPI
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

# Redirect HTTP to HTTPS
# Only enable when TLS is actually configured!
# app.add_middleware(HTTPSRedirectMiddleware)

# Prevent host header attacks
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["api.example.com", "*.example.com"],
)


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


# --- Option 2: Uvicorn with TLS ---
# Generate self-signed cert for testing:
# openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem \
#     -days 365 -nodes -subj '/CN=localhost'
#
# Run Uvicorn with TLS:
# uvicorn app.main:app \
#     --host 0.0.0.0 --port 443 \
#     --ssl-keyfile ./key.pem \
#     --ssl-certfile ./cert.pem


# --- Option 3: Let's Encrypt with certbot ---
# Install certbot:
# apt install certbot python3-certbot-nginx
#
# Obtain certificate:
# certbot --nginx -d api.example.com
#
# Auto-renewal (crontab):
# 0 0 * * 0 certbot renew --quiet


# --- Option 4: Caddy (automatic HTTPS) ---
# Caddyfile:
# api.example.com {
#     reverse_proxy localhost:8000
# }
# That's it! Caddy automatically obtains and renews Let's Encrypt certs.


# --- Nginx with Let's Encrypt ---
# /etc/nginx/sites-available/fastapi
NGINX_CONFIG = """
server {
    listen 80;
    server_name api.example.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_prefer_server_ciphers on;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;

    # HSTS
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload";

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
"""


# --- Security headers middleware ---
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["Strict-Transport-Security"] = (
            "max-age=63072000; includeSubDomains"
        )
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        return response

app.add_middleware(SecurityHeadersMiddleware)

Key Points

•Terminate TLS at the reverse proxy (Nginx, Caddy) or load balancer -- not in FastAPI directly
•Caddy provides the simplest HTTPS setup with automatic Let's Encrypt certificates
•Use HTTPSRedirectMiddleware to force HTTPS and TrustedHostMiddleware to prevent host header attacks
•Set HSTS headers to tell browsers to always use HTTPS for your domain

Gotchas

•Enabling HTTPSRedirectMiddleware without TLS configured creates an infinite redirect loop
•Self-signed certificates trigger browser warnings -- only use for development and testing
•Let's Encrypt certificates expire every 90 days -- always configure auto-renewal
•HSTS with preload is irreversible for the max-age duration -- start with a short max-age and increase gradually

Terminate TLS at the reverse proxy (Nginx + Let's Encrypt or Caddy's auto-HTTPS) rather than in FastAPI. Use HTTPSRedirectMiddleware to force HTTPS and add HSTS headers. For simple setups, Uvicorn supports --ssl-keyfile and --ssl-certfile directly.

•TLS at the proxy is standard; Caddy provides the easiest automatic HTTPS

# Caddy (easiest -- automatic HTTPS):
# api.example.com {
#     reverse_proxy localhost:8000
# }

# Uvicorn direct TLS:
# uvicorn app:app --ssl-keyfile key.pem --ssl-certfile cert.pem

# FastAPI middleware:
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)  # Only with TLS configured!

HTTPSRedirectMiddleware without TLS causes infinite redirects

Question 14 / 15

How do you use Docker Compose for FastAPI with dependencies?

Docker Compose orchestrates your FastAPI application along with its dependencies (database, Redis, reverse proxy) as a multi-container setup. It is ideal for local development and simple production deployments.

Typical service stack:

api: FastAPI application (built from Dockerfile)
db: PostgreSQL database
redis: Redis for caching/sessions
nginx: Reverse proxy for TLS and static files

Key Docker Compose features:

depends_on with condition: Wait for dependencies to be healthy before starting the app
volumes: Persist database data across container restarts
networks: Isolate services on an internal network
environment/env_file: Inject configuration without hardcoding
deploy.replicas: Run multiple API instances

Health check integration:

depends_on with condition: service_healthy ensures the API does not start until PostgreSQL is ready to accept connections.

Production vs development:

Development: Use docker compose watch for live reload
Production: Use docker compose -f docker-compose.yml -f docker-compose.prod.yml up

Pseudo Code

services:
  api:
    build: .
    depends_on:
      db: { condition: service_healthy }
      redis: { condition: service_healthy }
  db:
    image: postgres:16
    healthcheck: pg_isready
  redis:
    image: redis:7
  nginx:
    image: nginx
    ports: ["80:80", "443:443"]

Code

# --- docker-compose.yml ---
version: "3.9"  # Optional in modern Docker Compose

services:
  # --- FastAPI application ---
  api:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      - DATABASE_URL=postgresql+asyncpg://app:secret@db:5432/appdb
      - REDIS_URL=redis://redis:6379/0
      - ENVIRONMENT=production
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    deploy:
      replicas: 2  # Swarm only; use --scale for regular compose  # Run 2 instances
      restart_policy:
        condition: on-failure
        delay: 5s
        max_attempts: 3
    healthcheck:
      test: ["CMD", "python", "-c",
             "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
      interval: 30s
      timeout: 5s
      retries: 3
      start_period: 10s
    networks:
      - internal

  # --- PostgreSQL ---
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: appdb
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d appdb"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - internal

  # --- Redis ---
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 256mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks:
      - internal

  # --- Nginx reverse proxy ---
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/conf.d/default.conf:ro
      - ./nginx/certs:/etc/nginx/certs:ro
    depends_on:
      api:
        condition: service_healthy
    networks:
      - internal

volumes:
  postgres_data:
  redis_data:

networks:
  internal:
    driver: bridge


# --- Commands ---
# Start all services:
# docker compose up -d

# Scale API instances:
# docker compose up -d --scale api=4

# View logs:
# docker compose logs -f api

# Run migrations:
# docker compose exec api alembic upgrade head

# Stop and remove:
# docker compose down

# Stop and remove INCLUDING volumes (destroys data!):
# docker compose down -v

Key Points

•Use depends_on with condition: service_healthy to ensure the DB is ready before the API starts
•Persist database and Redis data with named volumes to survive container restarts
•Use deploy.replicas to run multiple API instances behind the Nginx reverse proxy
•Health checks on all services enable proper dependency ordering and monitoring

Gotchas

•depends_on only controls startup order -- it does not retry if the API fails to connect on first attempt
•docker compose down -v destroys all named volumes including database data -- use with extreme caution
•Container-to-container networking uses service names as hostnames (db, redis) -- not localhost
•Secrets in environment variables are visible in docker inspect -- use Docker secrets for production

Docker Compose runs FastAPI with PostgreSQL, Redis, and Nginx together. Use depends_on with condition: service_healthy for startup ordering, named volumes for data persistence, and deploy.replicas for scaling.

•depends_on + healthcheck ensures the DB is ready; named volumes persist data

# docker-compose.yml
services:
  api:
    build: .
    environment:
      - DATABASE_URL=postgresql+asyncpg://app:secret@db:5432/appdb
    depends_on:
      db: { condition: service_healthy }
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: appdb
    volumes: [postgres_data:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app"]
      interval: 10s
volumes:
  postgres_data:

docker compose down -v deletes all volumes including database data

Question 15 / 15

How do you implement zero-downtime deployments?

Zero-downtime deployment means updating your application without any period where requests are dropped or users see errors. This requires careful coordination between the load balancer, application instances, and database migrations.

Strategies:

Rolling update: Replace instances one at a time. While one instance is restarting, others continue serving traffic.
- Kubernetes default strategy
- Docker Compose with multiple replicas
- Gunicorn's kill -HUP for graceful reload
Blue/green deployment: Run two identical environments. Route traffic from the old (blue) to the new (green) after verification.
- Zero risk: instant rollback by switching back to blue
- Requires 2x resources during deployment
Canary deployment: Route a small percentage (e.g., 5%) of traffic to the new version. Gradually increase if metrics look good.
- Reduces blast radius of bad deployments
- Requires traffic routing capabilities (Istio, AWS ALB)

Prerequisites:

Health checks: Load balancer only routes to healthy instances
Graceful shutdown: Finish in-flight requests before stopping
Backward-compatible migrations: New code must work with both old and new schemas
Readiness probes: New instances report ready only after full initialization
Connection draining: Load balancer stops sending new requests to instances being shut down

Database migration safety:

Add columns as nullable first
Deploy code that works with both old and new schema
Run migration
Deploy code that uses the new column
This is called the expand-contract pattern

Pseudo Code

# Rolling update (Kubernetes)
spec:
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0  # Always keep all replicas available

# Blue/green (Nginx)
# Step 1: Deploy green on port 8001
# Step 2: Test green directly
# Step 3: Update Nginx upstream to 8001
# Step 4: Reload Nginx: nginx -s reload

Code

# --- Kubernetes rolling update manifest ---
# k8s/deployment.yaml
KUBERNETES_MANIFEST = """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1        # Add 1 new pod before removing old
      maxUnavailable: 0  # Never reduce below desired count
  selector:
    matchLabels:
      app: fastapi
  template:
    metadata:
      labels:
        app: fastapi
    spec:
      terminationGracePeriodSeconds: 30
      containers:
      - name: api
        image: myregistry/fastapi-app:v2
        ports:
        - containerPort: 8000
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8000
          initialDelaySeconds: 15
          periodSeconds: 20
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 5"]
"""


# --- Blue/green with Docker Compose + Nginx ---
# deploy.sh
DEPLOY_SCRIPT = """
#!/bin/bash
set -e

CURRENT=$(docker compose ps --format json | python3 -c "
import json, sys
for line in sys.stdin:
    svc = json.loads(line)
    if 'api-blue' in svc.get('Name', ''):
        print('blue')
        break
else:
    print('green')
")

if [ "$CURRENT" = "blue" ]; then
    NEW=green
    NEW_PORT=8001
else
    NEW=blue
    NEW_PORT=8000
fi

echo "Current: $CURRENT, Deploying: $NEW"

# Step 1: Start new version
docker compose up -d api-$NEW

# Step 2: Wait for health check
echo "Waiting for $NEW to be healthy..."
for i in $(seq 1 30); do
    if curl -sf http://localhost:$NEW_PORT/health/ready > /dev/null; then
        echo "$NEW is healthy!"
        break
    fi
    sleep 1
done

# Step 3: Switch Nginx to new version
cat > /etc/nginx/conf.d/upstream.conf << EOF
upstream api {
    server 127.0.0.1:$NEW_PORT;
}
EOF
nginx -s reload

# Step 4: Stop old version (after draining)
sleep 10  # Allow in-flight requests to complete
docker compose stop api-$CURRENT

echo "Deployment complete: $NEW is live"
"""


# --- FastAPI app with proper shutdown support ---
from contextlib import asynccontextmanager
from fastapi import FastAPI, Response
import asyncio
import logging

logger = logging.getLogger(__name__)

app_ready = False

@asynccontextmanager
async def lifespan(app: FastAPI):
    global app_ready
    # Startup: initialize resources
    logger.info("Starting application...")
    # await db.connect()
    # await cache.connect()
    app_ready = True
    logger.info("Application ready")

    yield

    # Shutdown: graceful cleanup
    app_ready = False
    logger.info("Draining connections...")
    await asyncio.sleep(2)  # Allow health checks to detect shutdown
    # await cache.close()
    # await db.close()
    logger.info("Shutdown complete")

app = FastAPI(lifespan=lifespan)

@app.get("/health/ready")
async def readiness(response: Response):
    """Returns 503 during startup and shutdown for proper traffic draining."""
    if not app_ready:
        response.status_code = 503
        return {"status": "not_ready"}
    return {"status": "ready"}

@app.get("/health/live")
async def liveness():
    return {"status": "alive"}

Key Points

•Rolling updates replace instances one at a time -- maxUnavailable: 0 ensures no capacity loss
•Readiness probes prevent traffic from reaching instances that are still starting or shutting down
•The preStop lifecycle hook (sleep 5) gives the load balancer time to stop routing before the app shuts down
•Blue/green provides instant rollback by switching back to the previous environment

Gotchas

•Database migrations must be backward-compatible -- both old and new code versions run simultaneously during rolling updates
•Without a preStop hook, Kubernetes may send traffic to a pod that is already shutting down (race condition with endpoint removal)
•Blue/green requires 2x resources during deployment -- cost considerations for large services
•Canary deployments need traffic splitting capabilities (Istio, Linkerd, or AWS ALB weighted target groups)

Use rolling updates (Kubernetes maxUnavailable: 0) or blue/green deployments with health checks. Readiness probes prevent traffic to starting/stopping instances. A preStop sleep gives the load balancer time to drain. Database migrations must be backward-compatible.

•maxUnavailable: 0 + readiness probes + preStop hook = zero-downtime rolling updates

# Kubernetes rolling update
spec:
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      containers:
      - name: api
        readinessProbe:
          httpGet: { path: /health/ready, port: 8000 }
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 5"]

Database migrations must work with both old and new code versions running simultaneously

Built by onllm.dev

Home GitHub Support awesomePrep

How do you configure Uvicorn for production?

Critical production flags:

Worker count formula:

Performance tuning:

Security:

How do you use Gunicorn with Uvicorn workers?

Why Gunicorn + Uvicorn:

Configuration:

Gunicorn config file (gunicorn.conf.py):

When to use Gunicorn:

When NOT to use Gunicorn:

How do you create a Dockerfile for FastAPI (using official python:3.x-slim image)?

Why NOT tiangolo/uvicorn-gunicorn-fastapi:

Best practices:

Image size comparison:

How do you manage environment variables and configuration?

Pydantic Settings features:

Priority order (highest to lowest):

Best practices:

How do you implement structured logging for FastAPI?

Why structured logging:

Popular libraries:

Key patterns:

Integration with Uvicorn:

How do you implement health check endpoints?

Types of health checks:

Kubernetes probe mapping:

Best practices:

How do you set up a reverse proxy (Nginx) for FastAPI?

Why use a reverse proxy:

Key Nginx directives for FastAPI:

WebSocket proxying:

Common gotcha:

How do you scale FastAPI applications horizontally?

Prerequisites for horizontal scaling:

Scaling strategies:

What breaks when you scale:

How do you handle graceful shutdown?

How it works in Uvicorn/FastAPI:

Key settings:

What to clean up during shutdown:

Signal handling:

How do you implement CI/CD for FastAPI applications?

Pipeline stages:

CI best practices for FastAPI:

Deployment strategies:

GitHub Actions is the most common CI/CD for open-source Python projects.

How do you monitor FastAPI applications in production?

1. Metrics (Prometheus):

2. Logs (structured):

3. Traces (OpenTelemetry):

The RED method (recommended for APIs):

Common alerting rules:

How do you handle database migrations in production?

Setup:

Production workflow:

Running migrations in Docker:

Key principles:

Safe migration patterns:

How do you implement HTTPS/TLS for FastAPI?

Three approaches:

Let's Encrypt (free certificates):

HTTPS redirect:

HSTS (HTTP Strict Transport Security):

Certificate types:

How do you use Docker Compose for FastAPI with dependencies?

Typical service stack:

Key Docker Compose features:

Health check integration:

Production vs development:

How do you implement zero-downtime deployments?

Strategies:

Prerequisites:

Database migration safety:

Deployment & Production

Gunicorn config file (`gunicorn.conf.py`):