Module 5: Capstone Project — API with Caching Strategy

Implementation: Caching + Rate Limiting

Next →

Overview

You build the first block of features on top of capsule 02's scaffolding. Here you implement the RateLimitMiddleware with a multi-tier sliding window (the 3 tiers free/pro/enterprise with differentiated limits), the CacheMiddleware that caches GET responses automatically, and the /api/products, /api/categories, and /api/users routers that use cache-aside with correct invalidation. By the end of the capsule, the API responds to real requests with working caching and rate limiting.

This capsule is 100% code. You saw the concepts in modules 2 and 3 — here you integrate them coherently. Every design decision applies the patterns you already learned: the rate limit middleware runs BEFORE the cache middleware (because we want to rate-limit even requests that would be a cache HIT), the cache middleware honors the endpoint's Cache-Control: max-age=, and the endpoints use consistent helpers for cache-aside. Capsule 04 adds sessions + Pub/Sub + write-through/write-behind on top of this foundation.

You'll verify the result with automated tests: 100 quick requests to confirm the rate limiting (the free tier rejects after 100 in an hour), hit rate benchmarks (cache-aside should reach >85% on realistic workloads), and correct HTTP headers on every response. It's the capsule where the project stops being scaffolding and starts being a real API.

The RateLimitMiddleware with a sliding window

Create app/rate_limit/sliding_window.py:

"""
A sliding window rate limiter (a refactor of module 3's capsule 03).
"""
import time
import uuid
import logging
from redis.asyncio import Redis
from redis.exceptions import RedisError


logger = logging.getLogger(__name__)


async def check_sliding_window(
    r: Redis,
    key: str,
    limit: int,
    window_seconds: int,
) -> tuple[bool, dict]:
    """
    Checks the rate limit with a sliding window.

    Returns:
        (allowed, info) where info includes:
        - limit, remaining, retry_after_seconds, reset_in_seconds
    """
    now = time.time()
    window_start = now - window_seconds
    member = f"req:{now}:{uuid.uuid4().hex[:8]}"

    try:
        pipe = r.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zadd(key, {member: now})
        pipe.zcard(key)
        pipe.expire(key, window_seconds + 60)
        _, _, count, _ = await pipe.execute()

        if count > limit:
            # Rollback
            await r.zrem(key, member)

            # Calculate retry_after from the oldest entry
            oldest = await r.zrange(key, 0, 0, withscores=True)
            if oldest:
                oldest_score = oldest[0][1]
                retry_after = max(0, oldest_score + window_seconds - now)
            else:
                retry_after = window_seconds

            return (False, {
                "limit": limit,
                "remaining": 0,
                "retry_after_seconds": round(retry_after, 1),
                "reset_in_seconds": int(window_seconds),
            })

        return (True, {
            "limit": limit,
            "remaining": limit - count,
            "retry_after_seconds": 0,
            "reset_in_seconds": int(window_seconds),
        })

    except RedisError as e:
        logger.warning(f"Rate limit check failed (Redis error): {e}. Allowing request (degraded).")
        return (True, {
            "limit": limit,
            "remaining": limit,
            "retry_after_seconds": 0,
            "reset_in_seconds": int(window_seconds),
            "degraded": True,
        })

app/rate_limit/middleware.py

"""
The RateLimitMiddleware for FastAPI.
"""
import time
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

from app.config import TIER_LIMITS, ENDPOINT_CATEGORY, DEFAULT_RATE_CATEGORY
from app.redis_client import get_redis
from app.rate_limit.sliding_window import check_sliding_window
from app.auth.jwt_handler import decode_jwt_safe   # capsule 04 creates it


logger = logging.getLogger(__name__)


SKIP_PATHS = {"/", "/health", "/metrics", "/docs", "/openapi.json", "/redoc"}


class RateLimitMiddleware(BaseHTTPMiddleware):
    """
    Multi-tier sliding window rate limiting.

    It identifies the user by:
    1. The JWT (if present) → the tier from the session in Redis
    2. An IP fallback → the "free" tier
    """

    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting on the system endpoints
        if request.url.path in SKIP_PATHS:
            return await call_next(request)

        # Identify the user and the tier
        user_id, tier = await self._get_user_and_tier(request)
        rate_key_id = f"user:{user_id}" if user_id else f"ip:{request.client.host if request.client else 'unknown'}"

        # Determine the endpoint's category
        endpoint_key = f"{request.method}:{request.url.path}"
        # Simplified path matching (in production, use a regex)
        category = self._match_category(request.method, request.url.path)

        # Get the tier's limit for this category
        tier_config = TIER_LIMITS.get(tier, TIER_LIMITS["free"])
        limit, window = tier_config[category]

        # Check the rate limit
        rate_key = f"rate:{tier}:{category}:{rate_key_id}"
        r = get_redis()

        allowed, info = await check_sliding_window(r, rate_key, limit, window)

        if not allowed:
            logger.info(f"Rate limited: {rate_key_id} on {category} (tier={tier})")
            return JSONResponse(
                status_code=429,
                content={
                    "error": "rate_limit_exceeded",
                    "message": f"Rate limit exceeded for {category}. Retry after {int(info['retry_after_seconds'])}s.",
                    "retry_after_seconds": int(info["retry_after_seconds"]),
                    "category": category,
                },
                headers={
                    "X-RateLimit-Limit": str(info["limit"]),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(int(time.time() + info["retry_after_seconds"])),
                    "Retry-After": str(int(info["retry_after_seconds"])),
                }
            )

        # Add headers and continue
        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(info["limit"])
        response.headers["X-RateLimit-Remaining"] = str(info["remaining"])
        response.headers["X-RateLimit-Reset"] = str(int(time.time() + info["reset_in_seconds"]))
        if info.get("degraded"):
            response.headers["X-RateLimit-Degraded"] = "true"

        return response

    async def _get_user_and_tier(self, request: Request) -> tuple[int | None, str]:
        """Extracts the user_id + tier from the JWT (if it exists)."""
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            return (None, "free")

        token = auth_header.split(" ", 1)[1]
        payload = decode_jwt_safe(token)
        if not payload:
            return (None, "free")

        # Check the session in Redis to get the current tier
        session_id = payload.get("session_id")
        if not session_id:
            return (payload.get("user_id"), "free")

        try:
            r = get_redis()
            tier = await r.hget(f"session:{session_id}", "tier")
            return (payload.get("user_id"), tier or "free")
        except Exception:
            return (payload.get("user_id"), "free")

    def _match_category(self, method: str, path: str) -> str:
        """Match endpoint to rate limit category."""
        # An exact match
        key = f"{method}:{path}"
        if key in ENDPOINT_CATEGORY:
            return ENDPOINT_CATEGORY[key]

        # A prefix match (for paths with an id)
        for endpoint_key, category in ENDPOINT_CATEGORY.items():
            ep_method, ep_path = endpoint_key.split(":", 1)
            if method == ep_method and path.startswith(ep_path):
                return category

        return DEFAULT_RATE_CATEGORY

The CacheMiddleware

Create app/caching/middleware.py:

"""
The CacheMiddleware: it caches GET responses automatically.
"""
import hashlib
import json
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
from redis.exceptions import RedisError

from app.redis_client import get_redis


logger = logging.getLogger(__name__)


CACHEABLE_PREFIXES = (
    "/api/products",
    "/api/categories",
    "/api/users",
)


SKIP_PATHS = {"/", "/health", "/metrics", "/docs", "/openapi.json", "/redoc"}


class CacheMiddleware(BaseHTTPMiddleware):
    """
    Caches GET responses with a TTL read from the Cache-Control header.

    It generates a cache key based on the path + sorted query params.
    It skips on a client's Cache-Control: no-cache.
    """

    def __init__(self, app, default_ttl: int = 300):
        super().__init__(app)
        self.default_ttl = default_ttl

    async def dispatch(self, request: Request, call_next):
        # Only cache GETs
        if request.method != "GET":
            return await call_next(request)

        # Skip the system endpoints
        if request.url.path in SKIP_PATHS:
            return await call_next(request)

        # Only cache the cacheable paths
        if not any(request.url.path.startswith(p) for p in CACHEABLE_PREFIXES):
            return await call_next(request)

        # Skip if the client asks for no-cache
        if "no-cache" in request.headers.get("cache-control", "").lower():
            return await call_next(request)

        cache_key = self._build_key(request)
        r = get_redis()

        # Try for a cache HIT
        try:
            cached = await r.get(cache_key)
            if cached:
                logger.debug(f"Cache HIT: {cache_key}")
                # Increment metrics
                await r.incr("metrics:cache:hits")
                return Response(
                    content=cached,
                    media_type="application/json",
                    headers={"X-Cache": "HIT"},
                )
        except RedisError as e:
            logger.warning(f"Cache read failed for {cache_key}: {e}")

        # Cache MISS: run the endpoint
        response = await call_next(request)
        try:
            await r.incr("metrics:cache:misses")
        except RedisError:
            pass

        # Cache it if it succeeded
        if response.status_code == 200:
            try:
                # Read the body
                body = b""
                async for chunk in response.body_iterator:
                    body += chunk

                # The TTL from the endpoint's Cache-Control, falling back to the default
                cache_control = response.headers.get("cache-control", "")
                ttl = self._extract_max_age(cache_control) or self.default_ttl

                await r.set(cache_key, body, ex=ttl)
                logger.debug(f"Cache MISS stored: {cache_key} (TTL={ttl}s)")

                # Rebuild the response (we consumed the body)
                return Response(
                    content=body,
                    status_code=200,
                    media_type=response.media_type,
                    headers={**dict(response.headers), "X-Cache": "MISS"},
                )
            except RedisError as e:
                logger.warning(f"Cache write failed for {cache_key}: {e}")
                return response

        return response

    def _build_key(self, request: Request) -> str:
        """Generates a unique cache key for the request."""
        # The path + sorted query params
        params = sorted(request.query_params.items())
        params_str = "&".join(f"{k}={v}" for k, v in params)
        raw = f"{request.url.path}?{params_str}"

        # Hash for short keys
        h = hashlib.sha256(raw.encode()).hexdigest()[:16]
        return f"cache:http:{request.url.path}:{h}"

    def _extract_max_age(self, cache_control: str) -> int | None:
        if "max-age=" in cache_control:
            try:
                return int(cache_control.split("max-age=")[1].split(",")[0].strip())
            except (IndexError, ValueError):
                pass
        return None

Routers: /api/products

Create app/routers/api.py:

"""
The protected API endpoints: products, categories, users (reads), search.
"""
import json
import logging
from fastapi import APIRouter, HTTPException, Depends, Response
from redis.asyncio import Redis

from app.config import CACHE_TTL, SEARCH_TOP_QUERIES, PUBSUB_INVALIDATE_PREFIX
from app.redis_client import get_redis_dep
from app.models import (
    Product, ProductCreate, ProductUpdate, ProductSummary,
    Category, User, UserUpdate
)
from app.db import (
    db_get_products, db_get_product, db_create_product,
    db_update_product, db_delete_product,
    db_get_categories, db_get_user, db_update_user,
)


logger = logging.getLogger(__name__)


router = APIRouter(prefix="/api", tags=["API"])


# ═══════════════════════════════════════════════════════════
# Products: cache-aside with invalidation
# ═══════════════════════════════════════════════════════════


@router.get("/products", response_model=list[ProductSummary])
async def list_products(
    response: Response,
    skip: int = 0,
    limit: int = 20,
    r: Redis = Depends(get_redis_dep),
):
    """
    Cache-aside with a 5 min TTL.
    The cache key includes skip+limit to distinguish the pages.
    The middleware already caches this automatically, but we also do it
    manually for the demo (in production, only one).
    """
    response.headers["Cache-Control"] = f"max-age={CACHE_TTL['products_list']}"

    # The middleware will cache it. The endpoint just returns the data.
    items = await db_get_products(skip=skip, limit=limit)
    return [ProductSummary(**p) for p in items]


@router.get("/products/{product_id}", response_model=Product)
async def get_product(
    product_id: int,
    response: Response,
    r: Redis = Depends(get_redis_dep),
):
    """Cache-aside with a 10 min TTL."""
    cache_key = f"cache:product:{product_id}"

    # Try the cache (manually, before the middleware)
    try:
        cached = await r.get(cache_key)
        if cached:
            response.headers["X-Cache"] = "HIT"
            return Product(**json.loads(cached))
    except Exception as e:
        logger.warning(f"Cache read failed: {e}")

    # A DB query
    product = await db_get_product(product_id)
    if not product:
        raise HTTPException(404, "Product not found")

    response.headers["Cache-Control"] = f"max-age={CACHE_TTL['product_detail']}"
    response.headers["X-Cache"] = "MISS"

    # Cache it (best effort)
    try:
        await r.set(cache_key, json.dumps(product), ex=CACHE_TTL["product_detail"])
    except Exception as e:
        logger.warning(f"Cache write failed: {e}")
        response.headers["X-Cache"] = "BYPASS"

    return Product(**product)


@router.post("/products", response_model=Product, status_code=201)
async def create_product(
    data: ProductCreate,
    r: Redis = Depends(get_redis_dep),
):
    """
    Create a product + invalidate the related caches.
    It also publishes an event on Pub/Sub.
    """
    new_product = await db_create_product(data.model_dump())

    # Invalidate caches manually
    try:
        # Invalidate the list caches (every page)
        cursor = 0
        while True:
            cursor, keys = await r.scan(cursor=cursor, match="cache:http:/api/products:*", count=100)
            if keys:
                await r.delete(*keys)
            if cursor == 0:
                break

        # Publish the invalidation event
        await r.publish(
            f"{PUBSUB_INVALIDATE_PREFIX}:product:list",
            json.dumps({"action": "create", "product_id": new_product["id"]})
        )
    except Exception as e:
        logger.warning(f"Cache invalidation failed: {e}")

    return Product(**new_product)


@router.put("/products/{product_id}", response_model=Product)
async def update_product(
    product_id: int,
    data: ProductUpdate,
    r: Redis = Depends(get_redis_dep),
):
    """An update + a specific invalidation + Pub/Sub."""
    updated = await db_update_product(product_id, data.model_dump(exclude_unset=True))
    if not updated:
        raise HTTPException(404, "Product not found")

    # Invalidate
    try:
        await r.delete(f"cache:product:{product_id}")
        # And the lists too
        cursor = 0
        while True:
            cursor, keys = await r.scan(cursor=cursor, match="cache:http:/api/products*", count=100)
            if keys:
                await r.delete(*keys)
            if cursor == 0:
                break

        # Pub/Sub
        await r.publish(
            f"{PUBSUB_INVALIDATE_PREFIX}:product:{product_id}",
            json.dumps({"action": "update", "product_id": product_id})
        )
    except Exception as e:
        logger.warning(f"Cache invalidation failed: {e}")

    return Product(**updated)


@router.delete("/products/{product_id}", status_code=204)
async def delete_product(
    product_id: int,
    r: Redis = Depends(get_redis_dep),
):
    """A delete + invalidation + Pub/Sub."""
    deleted = await db_delete_product(product_id)
    if not deleted:
        raise HTTPException(404, "Product not found")

    try:
        await r.delete(f"cache:product:{product_id}")
        cursor = 0
        while True:
            cursor, keys = await r.scan(cursor=cursor, match="cache:http:/api/products*", count=100)
            if keys:
                await r.delete(*keys)
            if cursor == 0:
                break

        await r.publish(
            f"{PUBSUB_INVALIDATE_PREFIX}:product:{product_id}",
            json.dumps({"action": "delete", "product_id": product_id})
        )
    except Exception as e:
        logger.warning(f"Cache invalidation failed: {e}")


# ═══════════════════════════════════════════════════════════
# Categories: cache-aside with a long TTL
# ═══════════════════════════════════════════════════════════


@router.get("/categories", response_model=list[Category])
async def list_categories(
    response: Response,
    r: Redis = Depends(get_redis_dep),
):
    """Cache-aside with a 1h TTL. They rarely change."""
    response.headers["Cache-Control"] = f"max-age={CACHE_TTL['categories']}"

    items = await db_get_categories()
    return [Category(**c) for c in items]


# ═══════════════════════════════════════════════════════════
# Users: cache-aside with a hash + a sliding TTL
# ═══════════════════════════════════════════════════════════


@router.get("/users/{user_id}", response_model=User)
async def get_user(
    user_id: int,
    response: Response,
    r: Redis = Depends(get_redis_dep),
):
    """
    Cache-aside with a hash. A sliding TTL renewed on every access.
    If the user is active, their profile stays in the cache.
    """
    cache_key = f"cache:user:{user_id}"

    # Try the cache (a hash)
    try:
        cached = await r.hgetall(cache_key)
        if cached:
            # Renew the TTL (sliding)
            await r.expire(cache_key, CACHE_TTL["user_profile"])
            response.headers["X-Cache"] = "HIT"
            return User(**cached)
    except Exception as e:
        logger.warning(f"Cache read failed: {e}")

    # The DB
    user = await db_get_user(user_id)
    if not user:
        raise HTTPException(404, "User not found")

    response.headers["X-Cache"] = "MISS"

    # Cache it (a hash)
    try:
        # Convert every value to a string for HSET
        cache_data = {k: str(v) for k, v in user.items()}
        await r.hset(cache_key, mapping=cache_data)
        await r.expire(cache_key, CACHE_TTL["user_profile"])
    except Exception as e:
        logger.warning(f"Cache write failed: {e}")
        response.headers["X-Cache"] = "BYPASS"

    return User(**user)


# The update endpoint (write-through) — capsule 04 adds it


# ═══════════════════════════════════════════════════════════
# Search: cache the top queries only
# ═══════════════════════════════════════════════════════════


@router.get("/search")
async def search(
    q: str,
    response: Response,
    r: Redis = Depends(get_redis_dep),
):
    """
    Selective cache-aside: it only caches queries in SEARCH_TOP_QUERIES.
    Unique/rare queries are NOT cached (an infinite space).
    """
    if not q:
        raise HTTPException(400, "Query parameter 'q' required")

    q_lower = q.lower().strip()

    # Only cache the top queries
    should_cache = q_lower in SEARCH_TOP_QUERIES
    cache_key = f"cache:search:{q_lower}" if should_cache else None

    # Try the cache
    if should_cache:
        try:
            cached = await r.get(cache_key)
            if cached:
                response.headers["X-Cache"] = "HIT"
                return json.loads(cached)
        except Exception as e:
            logger.warning(f"Cache read failed: {e}")

    # A mock search (in production: Elasticsearch, Algolia, etc.)
    import asyncio
    await asyncio.sleep(0.1)  # 100ms simulated search
    results = {
        "query": q,
        "results": [{"id": i, "name": f"Result for {q} #{i}"} for i in range(1, 11)],
        "total": 10,
    }

    response.headers["X-Cache"] = "MISS" if should_cache else "BYPASS"

    # Cache it if it applies
    if should_cache:
        try:
            await r.set(cache_key, json.dumps(results), ex=CACHE_TTL["search_top_queries"])
        except Exception as e:
            logger.warning(f"Cache write failed: {e}")

    return results

Integrating it in app/main.py

Modify app/main.py to add the middleware and the router:

"""
Production Cached API - the capsule 03 version (caching + rate limiting).
"""
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from redis.asyncio import Redis

from app.config import LOG_LEVEL, CORS_ORIGINS
from app.redis_client import init_pool, close_pool, get_redis, get_redis_dep
from app.db import seed_data

from app.rate_limit.middleware import RateLimitMiddleware
from app.caching.middleware import CacheMiddleware

from app.routers import api as api_router


logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s"
)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_pool()
    r = get_redis()
    try:
        await r.ping()
        logger.info("✓ Redis connected")
    except Exception as e:
        logger.error(f"✗ Redis failed: {e}")
        raise

    seed_data()
    logger.info("✓ Mock DB seeded")

    logger.info("✓ API ready")
    yield

    await close_pool()
    logger.info("✓ Cleanup complete")


app = FastAPI(
    title="Production Cached API",
    version="1.0.0",
    lifespan=lifespan,
)


# Middleware (added in reverse of execution order — the last one added runs first)
# Execution: CORS → RateLimit → Cache → endpoint

app.add_middleware(CacheMiddleware, default_ttl=300)
app.add_middleware(RateLimitMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Routers
app.include_router(api_router.router)


@app.get("/health")
async def health(r: Redis = Depends(get_redis_dep)):
    health_data = {"status": "healthy", "services": {}}
    try:
        await r.ping()
        info = await r.info("server")
        memory = await r.info("memory")
        health_data["services"]["redis"] = {
            "status": "healthy",
            "version": info.get("redis_version"),
            "memory_used": memory.get("used_memory_human"),
        }
    except Exception as e:
        health_data["status"] = "degraded"
        health_data["services"]["redis"] = {"status": "unhealthy", "reason": str(e)}
    return health_data


@app.get("/metrics")
async def metrics(r: Redis = Depends(get_redis_dep)):
    try:
        hits = int(await r.get("metrics:cache:hits") or 0)
        misses = int(await r.get("metrics:cache:misses") or 0)
        total = hits + misses
        hit_rate = (hits / total * 100) if total > 0 else 0

        return {
            "cache": {
                "hits": hits,
                "misses": misses,
                "hit_rate_percent": round(hit_rate, 2),
            },
        }
    except Exception as e:
        return {"error": str(e)}


@app.get("/")
async def root():
    return {"name": "Production Cached API", "docs": "/docs"}

⚠️ A temporary stub: decode_jwt_safe() in rate_limit/middleware.py — capsule 04 implements it. For now, create a stub:

# app/auth/jwt_handler.py (a temporary stub)
def decode_jwt_safe(token: str) -> dict | None:
    """STUB: capsule 04 implements it with PyJWT."""
    return None

Verification

# Start it
uvicorn app.main:app --reload

# Test 1: Health
curl http://localhost:8000/health | jq

# Test 2: GET products (cache MISS → HIT)
curl -i http://localhost:8000/api/products | head -10
# X-Cache: MISS
# X-RateLimit-Limit: 100   ← the free tier (no auth)
# X-RateLimit-Remaining: 99

curl -i http://localhost:8000/api/products | head -10
# X-Cache: HIT
# X-RateLimit-Remaining: 98   ← the rate limit counts EVERY request

# Test 3: GET with query params (a different cache key)
curl -i "http://localhost:8000/api/products?skip=10&limit=5" | head -10
# X-Cache: MISS (a different cache key)

curl -i "http://localhost:8000/api/products?skip=10&limit=5" | head -10
# X-Cache: HIT

# Test 4: GET a single product
curl -i http://localhost:8000/api/products/1 | head -10
# X-Cache: MISS (manual cache-aside)
curl -i http://localhost:8000/api/products/1 | head -10
# X-Cache: HIT

# Test 5: A POST invalidates the list
curl -X POST http://localhost:8000/api/products \
  -H "Content-Type: application/json" \
  -d '{"name": "New Product", "price": 99.99, "stock": 10, "category_id": 1}' | jq
# (it creates the product and invalidates the list cache)

curl -i http://localhost:8000/api/products | head -10
# X-Cache: MISS (the cache was invalidated)

# Test 6: A top search query (cached)
curl -i "http://localhost:8000/api/search?q=laptop" | head -10
# X-Cache: MISS
curl -i "http://localhost:8000/api/search?q=laptop" | head -10
# X-Cache: HIT

# Test 7: A rare search query (NOT cached)
curl -i "http://localhost:8000/api/search?q=xyz123" | head -10
# X-Cache: BYPASS (it doesn't cache)
curl -i "http://localhost:8000/api/search?q=xyz123" | head -10
# X-Cache: BYPASS (still not cached)

# Test 8: The rate limit (the free tier: 100/hr on general)
for i in {1..105}; do
  STATUS=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/api/products)
  if [ "$STATUS" != "200" ]; then echo "Req $i: $STATUS"; fi
done
# Expected: Req 101+ return a 429

# Test 9: Metrics
curl http://localhost:8000/metrics | jq
# {
#   "cache": {
#     "hits": 50,
#     "misses": 5,
#     "hit_rate_percent": 90.91
#   }
# }

Automated tests

Create tests/test_caching.py:

import pytest
import httpx


BASE = "http://localhost:8000"


@pytest.mark.asyncio
async def test_cache_hit_on_second_request():
    """Verify cache works: second request is HIT."""
    async with httpx.AsyncClient() as client:
        r1 = await client.get(f"{BASE}/api/categories")
        assert r1.status_code == 200
        # First might be MISS or HIT depending on previous tests

        r2 = await client.get(f"{BASE}/api/categories")
        assert r2.status_code == 200
        assert r2.headers["x-cache"] == "HIT"


@pytest.mark.asyncio
async def test_cache_invalidation_on_post():
    """Verify POST invalidates list cache."""
    async with httpx.AsyncClient() as client:
        # Prime cache
        await client.get(f"{BASE}/api/products?skip=0&limit=5")

        r_hit = await client.get(f"{BASE}/api/products?skip=0&limit=5")
        assert r_hit.headers["x-cache"] == "HIT"

        # POST invalidates
        await client.post(f"{BASE}/api/products", json={
            "name": "Test", "price": 10.0, "stock": 1, "category_id": 1,
        })

        # Next GET is MISS
        r_miss = await client.get(f"{BASE}/api/products?skip=0&limit=5")
        assert r_miss.headers["x-cache"] == "MISS"


@pytest.mark.asyncio
async def test_search_top_query_cached():
    async with httpx.AsyncClient() as client:
        r1 = await client.get(f"{BASE}/api/search?q=laptop")
        r2 = await client.get(f"{BASE}/api/search?q=laptop")
        assert r2.headers["x-cache"] == "HIT"


@pytest.mark.asyncio
async def test_search_rare_query_bypass():
    async with httpx.AsyncClient() as client:
        unique_q = f"random_{__import__('uuid').uuid4().hex[:8]}"
        r1 = await client.get(f"{BASE}/api/search?q={unique_q}")
        assert r1.headers["x-cache"] == "BYPASS"


@pytest.mark.asyncio
async def test_no_cache_header_skips():
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"{BASE}/api/products",
            headers={"Cache-Control": "no-cache"},
        )
        # The cache middleware honors no-cache. The endpoint can add X-Cache: MISS but the middleware must not serve a HIT.
        # In this implementation, when there's a no-cache, the middleware does NOT touch the response.
        # Verify the request went to the endpoint (it wasn't served from the cache)
# tests/test_rate_limit.py
import asyncio
import pytest
import httpx


BASE = "http://localhost:8000"


@pytest.mark.asyncio
async def test_rate_limit_kicks_in():
    """The free tier: 100/hr on general. Request 101 should be a 429."""
    async with httpx.AsyncClient() as client:
        # Clean up Redis first (in real tests)
        # ...

        # 100 requests OK
        for i in range(100):
            r = await client.get(f"{BASE}/api/products")
            assert r.status_code in (200, 429)  # Some might 429 if previous tests left counters

        # Eventually we hit the limit
        # Verify we see at least one 429
        r = await client.get(f"{BASE}/api/products")
        # If the previous 100 were in the same window, this one is a 429

Run them:

pytest tests/ -v -s

Troubleshooting

Problem 1: The middleware order is wrong

Cause: In FastAPI, middleware runs in reverse of the order you add it. The last one added runs first.

Solution:

# Execution: CORS → RateLimit → Cache → endpoint
# So you add them in reverse order:
app.add_middleware(CacheMiddleware)        # last → the first to run before the endpoint
app.add_middleware(RateLimitMiddleware)    # before that
app.add_middleware(CORSMiddleware)         # added first → the last to run

A reminder: on every request, the flow is: CORS takes the request → Rate Limit checks → the Cache looks → the Endpoint runs. On the response, it goes the other way.

Problem 2: The headers don't show up on 429 responses

Cause: The RateLimitMiddleware returns a JSONResponse(...) without the correct headers.

Solution: Make sure you include every required header in the JSONResponse(headers={...}).

Problem 3: The cache key includes the auth headers (a cache leak between users)

Cause: The cache key is based on the path + query params, NOT the headers. If two users make a GET /api/products, they share the cache key — this is CORRECT for public data.

Solution: For per-user data (e.g., /api/me/orders), do NOT use the automatic middleware. Cache it manually with the key cache:me:{user_id}:orders.

Problem 4: X-Cache: HIT but with old data

Cause: The invalidation pattern doesn't cover every cache key.

Solution: On every write operation, invalidate with a SCAN over a pattern:

async for key in r.scan_iter(match="cache:http:/api/products*"):
    await r.delete(key)

Problem 5: The rate limit counts requests served from the cache

Cause: The RateLimitMiddleware runs BEFORE the CacheMiddleware. This is deliberate.

Why: we want to rate-limit even users who only read from the cache. If an attacker makes 10000 requests/sec to /api/products (all cache HITs), it would still be bad for your app just from processing the middleware.

Solution: It's the correct behavior. Don't change it.

Problem 6: The SCAN pattern matching doesn't find every key

Cause: The middleware's cache key pattern includes a hash at the end.

Solution: Use a wildcard at the end:

# The cache keys: "cache:http:/api/products:abc123def456"
# The correct pattern:
await r.scan_iter(match="cache:http:/api/products*")  # with a wildcard

Summary

In this capsule you built:

The RateLimitMiddleware:

  • A sliding window with sorted sets (a refactor of M3's capsule 03)
  • Multi-tier: free/pro/enterprise with limits per category
  • It identifies the user by JWT (with a stub for capsule 04)
  • Standard HTTP headers: X-RateLimit-*, Retry-After
  • The 429 Too Many Requests status

The CacheMiddleware:

  • It caches GET responses automatically
  • The cache key: a hash of the path + sorted query params
  • It honors the endpoint's Cache-Control: max-age=
  • It skips on a client's Cache-Control: no-cache
  • Headers: X-Cache: HIT/MISS/BYPASS
  • Metrics: hit/miss counters

The /api/* routers:

  • /api/products: a list with automatic cache-aside
  • /api/products/{id}: the detail with manual cache-aside + a sliding TTL
  • POST /api/products: invalidates the list + a Pub/Sub event
  • PUT /api/products/{id}: a specific invalidation + Pub/Sub
  • DELETE /api/products/{id}: invalidation + Pub/Sub
  • /api/categories: a long cache TTL (1h)
  • /api/users/{id}: cache-aside with a HASH (granular updates)
  • /api/search: caches only the top queries (an infinite space)

Verification:

  • Correct headers in the responses
  • Working rate limiting (a 429 after N requests)
  • Cache hits/misses matching expectations
  • Correct invalidation (POST → the next GET is a MISS)
  • Working pattern matching for cacheable vs rare queries

The critical part: the API now has fully working caching + rate limiting. Everything you learned in M2 and M3 is professionally integrated. Capsule 04 adds sessions + a Pub/Sub listener + write-through/write-behind on top of this foundation.

Additional resources

  1. FastAPI Middleware Order — The ordering behavior
  2. Cache-Control HTTP Header — The header's spec
  3. Rate Limiting Best Practices — Stripe's patterns
  4. Cache Stampede Prevention — You saw it in M2's capsule 05; here you could add locking
  5. Async Python with Redis — The official patterns
  6. 12-Factor Logs — Why structured logging matters

What's next?

In Capsule 04 you complete the project by adding: JWT + Redis sessions (login, logout, logout-all, list devices), write-through for PUT /api/users/{id}, write-behind for POST /analytics/event with an async worker, a Pub/Sub listener that receives invalidation events and acts on them, a WebSocket bridge for real-time notifications, and extended metrics.

Keep Redis running. Let's go.

← PreviousArchitecture and Caching StrategyNext →Capsule 04 — Sessions + Pub/Sub + Monitoring