FastAPI Rate Limiting Demo — Complete Line‑by‑Line Documentation

This document walks a layperson through every file in the demo you’re building: a Dockerized FastAPI service that enforces per‑endpoint rate limits with API‑key tiers (free, pro, enterprise) and an optional OpenAI call (mockable). You’ll learn what each line does and why it exists.

Folder layout (from the previous message):

fastapi-rate-limit-demo/
├─ docker-compose.yml
├─ .env.example
├─ README.md
└─ app/
   ├─ Dockerfile
   ├─ requirements.txt
   ├─ config.py
   ├─ limiter.py
   ├─ tiers.py
   ├─ openai_client.py
   └─ main.py

1) docker-compose.yml — Orchestrates your containers

Purpose: Start two containers together:

File content with explanations:

version: "3.9"                # Compose file format version

services:                      # A set of containers that run together
  api:                         # The FastAPI web API service
    build: ./app               # Build Docker image from the ./app folder
    container_name: rate_limit_api  # Nice readable name for container
    env_file: .env             # Load environment variables from .env
    ports:
      - "8000:8000"           # Map host port 8000 → container port 8000
    depends_on:
      - redis                  # Ensure Redis starts before API
    restart: unless-stopped    # Auto‑restart unless you stop it

  redis:                       # The Redis service used by our limiter
    image: redis:7-alpine      # Lightweight Redis image
    container_name: rate_limit_redis
    command: ["redis-server", "--appendonly", "yes"]
                               # Enable AOF persistence (survives restarts)
    volumes:
      - redisdata:/data        # Store Redis data on a named volume
    ports:
      - "6379:6379"           # (Optional) expose Redis for local inspection
    restart: unless-stopped

volumes:
  redisdata:                   # Named volume (Docker manages the storage)

Mental model: Compose is like a movie director. It tells Docker “start these actors (containers), give them their props (env/volumes), and keep them on stage (restart).”

2) .env.example — Configuration you can edit safely

Purpose: Central place to change settings without touching code. You’ll copy it to .env and edit values.

File content with explanations:

# FastAPI app
APP_ENV=dev                   # Environment name (dev, staging, prod)
APP_HOST=0.0.0.0              # Host address FastAPI binds to inside container
APP_PORT=8000                 # Port number inside container

# Redis
REDIS_URL=redis://redis:6379/0  # Connect to the 'redis' service db 0

# OpenAI (set to a real key or leave mock mode on)
OPENAI_API_KEY=               # If empty or mock mode is 1, we don't call OpenAI
USE_MOCK_OPENAI=1             # 1 = mock responses; 0 = call OpenAI API

# Default per-tier limits (requests per window per endpoint unless overridden)
# Format: TIER:requests,window_seconds
TIER_DEFAULTS=free:10,60;pro:60,60;enterprise:300,60

# Demo API keys mapped to tiers (for PoC)
# Format: key1:tier;key2:tier;...
API_KEYS=FREEDEMO123:free;PRODEMO456:pro;ENTDEMO789:enterprise

# Optional per-endpoint overrides
# Format: endpoint|tier=limit,window;endpoint|tier=limit,window;...
ENDPOINT_OVERRIDES=chat|free=5,60;chat|pro=30,60;chat|enterprise=180,60

Why this matters: Product teams can adjust limits and keys without redeploying code.

3) app/Dockerfile — How to build your API’s image

Purpose: Describe the recipe to create a runnable container image for your FastAPI app.

File content with explanations:

FROM python:3.11-slim
# Use a small official Python image for faster builds and smaller size

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1
# Don’t create .pyc files, and flush logs immediately (useful in containers)

WORKDIR /app
# All following commands run from /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    curl build-essential \
  && rm -rf /var/lib/apt/lists/*
# Install tiny set of system packages (curl; build tools for possible deps)
# Clean apt cache to keep image small

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Install Python dependencies first to leverage Docker layer caching

COPY . .
# Copy the rest of the app into the image

EXPOSE 8000
# Document that the app listens on port 8000 (helps tools, not a firewall)

# IMPORTANT: use sh -c so ${APP_HOST}/${APP_PORT} get expanded from env
CMD ["sh","-c","uvicorn main:app --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000}"]

Key lesson: The CMD uses sh -c so environment variable defaults work inside the container. (Your earlier error came from not expanding ${APP_PORT}.)

4) app/requirements.txt — Python packages

Purpose: Exact libraries your app needs.

fastapi==0.114.2           # The web framework
uvicorn[standard]==0.30.6  # ASGI server to run FastAPI
redis==5.0.8               # Redis client (for rate limiting storage)
pydantic==2.9.2            # Config/validation utilities
httpx==0.27.2              # Async HTTP client (for OpenAI calls)
python-dotenv==1.0.1       # Load .env into environment variables

Why pinned versions? Reproducibility. Your demo behaves the same everywhere.

5) app/config.py — Centralized settings loader

Purpose: Read environment variables once and expose a typed settings object to the rest of the app.

Code (with commentary):

from pydantic import BaseModel   # Pydantic gives us typed, validated models
import os                        # Access environment variables

class Settings(BaseModel):
    app_env: str = os.getenv("APP_ENV", "dev")          # "dev" if not set
    host: str = os.getenv("APP_HOST", "0.0.0.0")        # default listen addr
    port: int = int(os.getenv("APP_PORT", "8000"))       # convert to int

    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")

    openai_api_key: str | None = os.getenv("OPENAI_API_KEY")
    use_mock_openai: bool = os.getenv("USE_MOCK_OPENAI", "1") == "1"

    # Tier defaults: "free:10,60;pro:60,60"
    tier_defaults_raw: str = os.getenv(
        "TIER_DEFAULTS", "free:10,60;pro:60,60;enterprise:300,60"
    )

    # API keys mapping: "FREEDEMO123:free;PRODEMO456:pro"
    api_keys_raw: str = os.getenv(
        "API_KEYS", "FREEDEMO123:free;PRODEMO456:pro;ENTDEMO789:enterprise"
    )

    # Endpoint overrides: "chat|free=5,60;chat|pro=30,60"
    endpoint_overrides_raw: str = os.getenv("ENDPOINT_OVERRIDES", "")

settings = Settings()  # Create a single, shared settings instance

Plain‑English summary: Every other file imports settings so they don’t juggle environment variables themselves.

6) app/tiers.py — Parse tiers, keys, and overrides

Purpose: Convert human‑readable strings in .env into Python dictionaries.

Code (with commentary):

from __future__ import annotations            # enables modern type hints
from typing import Dict, Tuple
from config import settings                   # import the shared settings

# "free:10,60;pro:60,60" -> {"free": (10,60), "pro": (60,60)}
def parse_tier_defaults(raw: str) -> Dict[str, Tuple[int, int]]:
    tiers = {}
    if not raw.strip():
        return tiers
    for part in raw.split(";"):
        name, spec = part.split(":")            # split at the ':'
        limit, window = spec.split(",")          # split "10,60" into 10 and 60
        tiers[name.strip()] = (int(limit), int(window))
    return tiers

# "FREEDEMO123:free;PRODEMO456:pro" -> {"FREEDEMO123":"free", ...}
def parse_api_keys(raw: str) -> Dict[str, str]:
    m = {}
    if not raw.strip():
        return m
    for part in raw.split(";"):
        key, tier = part.split(":")
        m[key.strip()] = tier.strip()
    return m

# "chat|free=5,60;chat|pro=30,60" -> {"chat": {"free": (5,60), ...}}
def parse_overrides(raw: str) -> Dict[str, Dict[str, Tuple[int, int]]]:
    out: Dict[str, Dict[str, Tuple[int,int]]] = {}
    if not raw.strip():
        return out
    for part in raw.split(";"):
        scope, spec = part.split("=")              # left: chat|free, right: 5,60
        endpoint, tier = scope.split("|")           # endpoint = chat, tier = free
        limit, window = spec.split(",")
        out.setdefault(endpoint.strip(), {})[tier.strip()] = (int(limit), int(window))
    return out

# Preparsed dictionaries available to the rest of the app
TIER_DEFAULTS = parse_tier_defaults(settings.tier_defaults_raw)
API_KEYS = parse_api_keys(settings.api_keys_raw)
ENDPOINT_OVERRIDES = parse_overrides(settings.endpoint_overrides_raw)

# Return the tier name for a given API key string
def get_tier_for_key(api_key: str | None) -> str | None:
    if not api_key:
        return None
    return API_KEYS.get(api_key)

# Return (limit, window_seconds) for an endpoint+tier pair
def get_limits_for(endpoint_name: str, tier: str) -> tuple[int, int]:
    # First check per‑endpoint overrides
    if endpoint_name in ENDPOINT_OVERRIDES and tier in ENDPOINT_OVERRIDES[endpoint_name]:
        return ENDPOINT_OVERRIDES[endpoint_name][tier]
    # Otherwise fall back to tier defaults
    if tier in TIER_DEFAULTS:
        return TIER_DEFAULTS[tier]
    # Unknown tier: be conservative
    return (1, 60)

Takeaway: This file translates friendly strings into actionable numbers so the limiter knows how many requests to allow.

7) app/limiter.py — The Redis‑backed rate limiter

Purpose: Enforce a fixed‑window rate limit per key and endpoint, and attach standard headers to responses.

Code (with commentary):

from __future__ import annotations
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
import time
import redis
from config import settings

# Create a global Redis client using the URL from .env
r = redis.from_url(settings.redis_url, decode_responses=True)

# Build a unique key and compute time left in the current window
def window_key(api_key: str, endpoint: str, window_seconds: int) -> tuple[str, int]:
    now = int(time.time())
    window_start = now - (now % window_seconds)  # align windows: 0–59, 60–119, etc.
    window_end = window_start + window_seconds
    remaining = max(0, window_end - now)        # seconds until the window resets
    key = f"rl:{api_key}:{endpoint}:{window_start}:{window_seconds}"
    return key, remaining

# The main function you call before serving an endpoint response
async def enforce_rate_limit(request: Request, *, api_key: str, endpoint: str, limit: int, window_seconds: int):
    k, ttl = window_key(api_key, endpoint, window_seconds)

    # Use a pipeline to bundle Redis commands efficiently
    pipe = r.pipeline()
    pipe.incr(k, amount=1)   # 1) increment the counter for this key
    pipe.ttl(k)              # 2) check if it already has a TTL
    current_count, current_ttl = pipe.execute()

    if current_ttl == -1:    # -1 => key exists but no expiry set yet
        r.expire(k, ttl)     # set it to expire when the window ends

    remaining_requests = max(0, limit - current_count)

    # Stash details on the request so middleware can write headers later
    request.state.rate_limit = {
        "limit": limit,
        "remaining": remaining_requests if remaining_requests > 0 else 0,
        "reset": ttl,                  # seconds until next window
        "window": window_seconds
    }

    if current_count > limit:
        # Over the limit: raise 429 and include helpful headers
        headers = {
            "X-RateLimit-Limit": str(limit),
            "X-RateLimit-Remaining": str(0),
            "X-RateLimit-Reset": str(ttl),
            "Retry-After": str(ttl),
        }
        raise HTTPException(status_code=429, detail="Rate limit exceeded", headers=headers)

# Helper that middleware calls to append headers if rate limit data was set
def rate_limit_headers_middleware(response, request):
    data = getattr(request.state, "rate_limit", None)
    if not data:
        return
    response.headers["X-RateLimit-Limit"] = str(data["limit"])
    response.headers["X-RateLimit-Remaining"] = str(data["remaining"])
    response.headers["X-RateLimit-Reset"] = str(data["reset"])

Plain‑English: We count requests in Redis, bucketed by time windows. If you exceed your allowance, you get a 429 “too many requests,” and the response tells you when to try again.

8) app/openai_client.py — Real or mocked OpenAI calls

Purpose: Demonstrate a real external call (OpenAI) while keeping the demo runnable without an API key.

Code (with commentary):

from __future__ import annotations
import os
from config import settings
import httpx

OPENAI_API_BASE = "https://api.openai.com/v1"  # Base URL for OpenAI API

# Single function our endpoint uses to get a reply
async def chat_complete(user_prompt: str) -> str:
    if settings.use_mock_openai or not settings.openai_api_key:
        # Safe, offline‑friendly path
        return f"[MOCKED] You said: {user_prompt[:80]}..."

    headers = {
        "Authorization": f"Bearer {settings.openai_api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": "gpt-4o-mini",
        "messages": [
            {"role":"system", "content": "You are a concise assistant."},
            {"role":"user", "content": user_prompt}
        ],
        "temperature": 0.2,
    }
    # Make an async HTTP POST to OpenAI
    async with httpx.AsyncClient(timeout=30) as client:
        res = await client.post(f"{OPENAI_API_BASE}/chat/completions", headers=headers, json=payload)
        res.raise_for_status()                  # raise error if not 2xx
        data = res.json()
        return data["choices"][0]["message"]["content"].strip()

Why mock? So your demo never fails in classrooms or on laptops without internet.

9) app/main.py — The FastAPI application

Purpose: Define routes (URLs) and glue everything together.

Code (with commentary):

from __future__ import annotations
from fastapi import FastAPI, Request, Depends, Header, HTTPException
from fastapi.responses import JSONResponse
from typing import Optional
from limiter import enforce_rate_limit, rate_limit_headers_middleware
from tiers import get_tier_for_key, get_limits_for
from openai_client import chat_complete

app = FastAPI(title="Endpoint Rate Limit Demo", version="1.0.0")

# Dependency: validate API key and figure out the user's tier
async def require_api_key(x_api_key: Optional[str] = Header(default=None, alias="X-API-Key")) -> tuple[str, str]:
    tier = get_tier_for_key(x_api_key or "")
    if not x_api_key or not tier:
        # For this demo, no/invalid key → 401 Unauthorized
        # (You could choose to treat no key as "free" if you prefer.)
        raise HTTPException(status_code=401, detail="Invalid or missing API key")
    return x_api_key, tier

# Middleware runs on every request/response
@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
    response = await call_next(request)                       # let route run
    rate_limit_headers_middleware(response, request)          # then add headers
    return response

# Health check route (no auth) — useful for Docker/monitors
@app.get("/health")
async def health():
    return {"status": "ok"}

# Simple route to show which key/tier you are
@app.get("/whoami")
async def whoami(api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    return {"api_key": api_key, "tier": tier}

# Public info route (no auth, no limits)
@app.get("/public-info")
async def public_info():
    return {"message": "This route is fully public, no limits."}

# Example POST that calls OpenAI (or mock) — rate limited per tier
@app.post("/chat")
async def chat(payload: dict, request: Request, api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    endpoint_name = "chat"
    limit, window = get_limits_for(endpoint_name, tier)   # e.g., free=5/60s

    # Enforce BEFORE doing work
    await enforce_rate_limit(request, api_key=api_key, endpoint=endpoint_name, limit=limit, window_seconds=window)

    user_prompt = (payload or {}).get("prompt", "")
    if not user_prompt:
        raise HTTPException(status_code=422, detail="Missing 'prompt' in body")

    answer = await chat_complete(user_prompt)
    return {"tier": tier, "answer": answer, "limit": limit, "window_seconds": window}

# Another example endpoint to show different limits per endpoint
@app.post("/ingest")
async def ingest(payload: dict, request: Request, api: tuple[str,str] = Depends(require_api_key)):
    api_key, tier = api
    endpoint_name = "ingest"
    limit, window = get_limits_for(endpoint_name, tier)    # default if no override
    await enforce_rate_limit(request, api_key=api_key, endpoint=endpoint_name, limit=limit, window_seconds=window)

    item = (payload or {}).get("item", {})                 # pretend to store it
    return {"stored": True, "item": item, "tier": tier, "limit": limit, "window_seconds": window}

Request path (what happens when you call /chat):

  1. Middleware runs later to add headers — it waits for the route to finish.
  2. require_api_key reads X-API-Key, finds your tier (free, pro, …).
  3. Route fetches limits for (endpoint='chat', tier).
  4. enforce_rate_limit increments Redis counter, possibly throws 429.
  5. If allowed, we call OpenAI (or mock) and return the result.
  6. Middleware appends X-RateLimit-* headers so clients know their status.

10) README.md — How to use the project (human manual)

Purpose: Make it easy for someone to clone, configure, run, and test the demo.

Sections to include (already provided in the template):

How to Run — Step‑by‑Step

  1. Create files exactly as shown above.
  2. Copy .env.example → .env and (optionally) set a real OPENAI_API_KEY.
  3. In the project root, run:
docker compose up --build

  1. Open docs in your browser:

  1. Try public route:
curl -s http://localhost:8000/public-info | jq
  1. Try authenticated routes:
curl -s -H "X-API-Key: FREEDEMO123" http://localhost:8000/whoami | jq
  1. See rate limiting in action (free is strict on /chat):
for i in {1..7}; do
  echo "Request $i";
  curl -s -H "X-API-Key: FREEDEMO123" -H "Content-Type: application/json" \
    -d '{"prompt":"Say hi"}' http://localhost:8000/chat -i | sed -n '1,20p';
  echo; echo;
done

Observe the headers:

Visual: Flow of a /chat Request

Client → FastAPI (/chat)
  └─ require_api_key() → tier (free/pro/enterprise)
       └─ get_limits_for("chat", tier)
            └─ enforce_rate_limit() → Redis counter+TTL
                 ├─ under limit → call chat_complete() → OpenAI/mock → response
                 └─ over limit  → raise 429 with Retry-After
  └─ middleware attaches X-RateLimit-* headers
Client receives JSON + headers

Customization Guide

A) Treat missing API key as free tier

if not x_api_key:
    return ("ANONYMOUS", "free")

…and make sure TIER_DEFAULTS has a free entry.

B) Different rules per endpoint

ENDPOINT_OVERRIDES=chat|free=5,60;ingest|free=2,60

C) Token Bucket (smoother bursting)

D) Mint API keys dynamically

E) Observability

Troubleshooting

FAQ (Layman‑friendly)

Q: What is rate limiting? A speed limit for API requests. It prevents abuse and lets you sell higher speed to paying customers.

Q: Why Redis? It’s super fast and can be shared across many app servers, so all instances enforce the same limits.

Q: What’s an API key? A secret string your client sends (here via X-API-Key header). It tells the server who you are and which tier you belong to.

Q: What are the X-RateLimit-* headers? They tell clients how many requests they have left and when the counter will reset.

Q: What’s the difference between fixed window and token bucket? Fixed window is easier to understand (counts reset every minute). Token bucket allows small bursts and is smoother under load.

Next Steps (Optional Enhancements)

With this guide, a non‑engineer can understand why each file exists and what each line contributes to the overall behavior. Feel free to paste this into your repo as DOCS.md and iterate. Enjoy showcasing free vs paid rate limits with confidence!