Architecture and Caching Strategy

Overview

This capsule lays the Production Cached API's foundations: the CACHING-STRATEGY.md document that decides which pattern to use for each endpoint, the Pydantic models that define the shape of the data, the Redis singleton client the whole app shares, and the FastAPI scaffolding with a lifespan + middleware stack. Without this foundation, the code in the following capsules has nothing to rest on.

The caching strategy document is a deliverable just as important as the code. On a real team, this document explains to other developers what gets cached, why, with which TTL, and how it's invalidated. Without documentation, the caching strategy is tacit knowledge that dies when the developer moves to another project. You'll learn to write it professionally — it's one of the portfolio criteria that separates a serious project from a tutorial project.

Then you build the technical scaffolding: app/config.py with centralized configuration, app/redis_client.py with the pool's singleton, app/models.py with the Pydantic models, and app/main.py with the FastAPI app + lifespan + middleware stack. Each file is a reference — you copy the structure and understand why each decision is where it is. Capsule 03 will fill in the routers; capsule 04 will complete them with sessions + Pub/Sub.

CACHING-STRATEGY.md

This document is the first deliverable. Create CACHING-STRATEGY.md at the root of the project:

# Caching Strategy — Production Cached API

## Philosophy

This API has 3 priorities, in order:

1. **Consistency where it matters:** critical data is always fresh
2. **Performance where it matters:** frequent reads are ultra-fast
3. **Resilience always:** the API works even if Redis fails (degraded mode)

## Patterns per endpoint

### Reads (cache-aside is the default)

| Endpoint | Pattern | TTL | Invalidation | Reason |
|----------|---------|-----|--------------|-------|
| `GET /api/products` | Cache-aside | 5min | DELETE on POST/PUT/DELETE | Massive reads, infrequent changes |
| `GET /api/products/{id}` | Cache-aside | 10min | DELETE + Pub/Sub | Same reasoning, a specific key |
| `GET /api/categories` | Cache-aside | 1h | Manual on update | Rare changes, heavy reads |
| `GET /api/users/{id}` | Cache-aside with a hash | 30min sliding | DELETE on PUT | A hash allows granular updates |
| `GET /api/search` | Cache-aside (top queries) | 2min | The natural TTL | An infinite space, only cache the common ones |

### Writes (the decision depends on context)

| Endpoint | Pattern | Reason |
|----------|---------|-------|
| `POST /api/products` (admin) | Cache-aside (invalidate on write) | The change is visible immediately, atomicity matters |
| `PUT /api/users/{id}` | Write-through | The UX requires immediate consistency (seeing changes reflected instantly) |
| `POST /api/orders` | NO cache | A critical transaction with a stock check + payment |
| `POST /analytics/event` | Write-behind | Massive volume, losing some events is acceptable |

## TTL Rationale

| TTL | Context |
|-----|----------|
| 2 min | Volatile data (popular searches) |
| 5 min | Read-heavy data with occasional writes (the product list) |
| 10 min | Data specific to an ID with stale tolerance (the product detail) |
| 30 min sliding | Active user data (the profile) |
| 1 hr | Near-static configuration (categories) |

## Invalidation

3 strategies in use:

1. **DELETE on write (manual):** most endpoints. When the admin endpoint updates, it deletes the local cache
2. **Event-driven (Pub/Sub):** for distributed systems. POST `/api/products/{id}` publishes `cache:invalidate:product:{id}`. Other subscribed services invalidate their caches
3. **The natural TTL (passive):** for data where stale is fine (search, analytics)

## Rate Limiting

A sliding window with sorted sets (ZADD with a timestamp, ZREMRANGEBYSCORE for cleanup, ZCARD to count). Differentiated categories:

- general (products, categories): 100/1000/10000 req/hr per tier
- search: 20/200/2000 req/hr (more expensive)
- orders: 10/100/1000 req/hr (critical)
- analytics: 1000/10000/100000 req/hr (high volume, batched)

## Sessions

JWT + Redis sessions. The JWT verifies identity (stateless, fast). The Redis session allows immediate revocation + mutable state (roles).

- A sliding TTL of 30min (maintained while the user is active)
- An absolute TTL of 24h maximum
- Local logout: revoke one session
- Global logout: revoke all of the user's sessions
- A password change: revoke all + create a new one for the current device

## Graceful Degradation

If Redis fails:
- Cache reads → straight to the DB (latency rises, the app works)
- Cache writes → skipped silently with a warning log
- Rate limiting → allow all (better to allow temporary spam than to reject legitimate users)
- Sessions → JWT only (no revocation check, a controlled risk)
- Pub/Sub publish → skipped silently
- The WebSocket bridge → disconnected, but the clients reconnect

The health endpoint reports `degraded` (no crash).

## Metrics

A hits/misses tracker per endpoint. The goal: a hit rate >80% on cached endpoints.

## Known limitations

- The data is in memory since we simulate the DB. Real production uses PostgreSQL.
- No distributed cluster. If you need multi-region, consider Redis Cluster.
- Pub/Sub isn't durable. Messages are lost if the subscriber is down.

---

**Last updated:** 2026-04-25
**Version:** 1.0
**Maintained by:** Mike Nieva

This document is updated whenever you add/modify endpoints. It's part of code review.

app/config.py

"""
The Production Cached API's centralized configuration.
"""
import os


# ═══════════════════════════════════════════════════════════
# Redis
# ═══════════════════════════════════════════════════════════

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
REDIS_MAX_CONNECTIONS = int(os.getenv("REDIS_MAX_CONNECTIONS", "100"))
REDIS_SOCKET_TIMEOUT = float(os.getenv("REDIS_SOCKET_TIMEOUT", "2.0"))


# ═══════════════════════════════════════════════════════════
# JWT + Sessions
# ═══════════════════════════════════════════════════════════

JWT_SECRET = os.getenv("JWT_SECRET", "change-me-in-production-please")
JWT_ALGORITHM = "HS256"
JWT_TTL_HOURS = 24

SESSION_TTL_SECONDS = 30 * 60          # 30 min sliding
SESSION_ABSOLUTE_TTL_SECONDS = 24 * 3600  # 24h maximum
MAX_SESSIONS_PER_USER = 5


# ═══════════════════════════════════════════════════════════
# Caching TTLs (in seconds)
# ═══════════════════════════════════════════════════════════

CACHE_TTL = {
    "products_list": 300,        # 5 min
    "product_detail": 600,       # 10 min
    "categories": 3600,          # 1 hour
    "user_profile": 1800,        # 30 min (sliding)
    "search_top_queries": 120,   # 2 min
}

# The top search queries to cache (in production, this would come from analytics)
SEARCH_TOP_QUERIES = {
    "laptop", "phone", "tablet", "shoes", "book",
    "headphones", "watch", "camera", "gaming", "monitor"
}


# ═══════════════════════════════════════════════════════════
# The rate limit per tier (a sliding window: limit + window_seconds)
# ═══════════════════════════════════════════════════════════

TIER_LIMITS = {
    "free": {
        "general":   (100, 3600),
        "search":    (20, 3600),
        "orders":    (10, 3600),
        "analytics": (1000, 3600),
    },
    "pro": {
        "general":   (1000, 3600),
        "search":    (200, 3600),
        "orders":    (100, 3600),
        "analytics": (10000, 3600),
    },
    "enterprise": {
        "general":   (10000, 3600),
        "search":    (2000, 3600),
        "orders":    (1000, 3600),
        "analytics": (100000, 3600),
    },
}

# The endpoint → rate limit category mapping
ENDPOINT_CATEGORY = {
    "GET:/api/products":          "general",
    "GET:/api/categories":        "general",
    "GET:/api/users":             "general",
    "PUT:/api/users":             "general",
    "POST:/api/products":         "general",
    "GET:/api/search":            "search",
    "POST:/api/orders":           "orders",
    "POST:/analytics/event":      "analytics",
}
DEFAULT_RATE_CATEGORY = "general"


# ═══════════════════════════════════════════════════════════
# Pub/Sub
# ═══════════════════════════════════════════════════════════

PUBSUB_INVALIDATE_PREFIX = "cache:invalidate"
PUBSUB_NOTIFICATIONS_PREFIX = "notifications"


# ═══════════════════════════════════════════════════════════
# Write-behind (analytics)
# ═══════════════════════════════════════════════════════════

ANALYTICS_BUFFER_KEY = "wb:analytics:events"
ANALYTICS_FLUSH_INTERVAL = 5     # seconds
ANALYTICS_FLUSH_BATCH_SIZE = 100


# ═══════════════════════════════════════════════════════════
# CORS / Misc
# ═══════════════════════════════════════════════════════════

CORS_ORIGINS = os.getenv("CORS_ORIGINS", "*").split(",")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")

app/redis_client.py

"""
A singleton for the async Redis client.
"""
import logging
from redis.asyncio import Redis, ConnectionPool

from app.config import REDIS_URL, REDIS_MAX_CONNECTIONS, REDIS_SOCKET_TIMEOUT


logger = logging.getLogger(__name__)


_pool: ConnectionPool | None = None
_client: Redis | None = None


def init_pool():
    """Initializes the global pool. Call it ONCE in the lifespan startup."""
    global _pool, _client

    _pool = ConnectionPool.from_url(
        REDIS_URL,
        max_connections=REDIS_MAX_CONNECTIONS,
        decode_responses=True,
        socket_timeout=REDIS_SOCKET_TIMEOUT,
        socket_connect_timeout=REDIS_SOCKET_TIMEOUT,
        socket_keepalive=True,
        health_check_interval=30,
    )

    _client = Redis(connection_pool=_pool)
    logger.info(f"Redis pool initialized: max_connections={REDIS_MAX_CONNECTIONS}")


def get_redis() -> Redis:
    """Returns the Redis client. It raises if it isn't initialized."""
    if _client is None:
        raise RuntimeError("Redis pool not initialized. Did you forget init_pool() in lifespan?")
    return _client


async def close_pool():
    """Closes the pool. Call it in the lifespan shutdown."""
    global _pool, _client
    if _client:
        await _client.aclose()
        _client = None
    if _pool:
        await _pool.aclose()
        _pool = None
    logger.info("Redis pool closed")


# The dependency for FastAPI
async def get_redis_dep() -> Redis:
    """A FastAPI dependency: use it with Depends(get_redis_dep)."""
    return get_redis()

app/models.py

"""
The Production Cached API's Pydantic models.
"""
from enum import Enum
from datetime import datetime
from pydantic import BaseModel, Field


# ═══════════════════════════════════════════════════════════
# Enums
# ═══════════════════════════════════════════════════════════


class Tier(str, Enum):
    free = "free"
    pro = "pro"
    enterprise = "enterprise"


class OrderStatus(str, Enum):
    pending = "pending"
    confirmed = "confirmed"
    shipped = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"


# ═══════════════════════════════════════════════════════════
# Auth
# ═══════════════════════════════════════════════════════════


class LoginRequest(BaseModel):
    username: str
    password: str


class LoginResponse(BaseModel):
    access_token: str
    session_id: str
    user_id: int
    tier: Tier


class SessionInfo(BaseModel):
    session_id: str
    device_name: str
    ip: str
    last_active: str
    is_current: bool = False


# ═══════════════════════════════════════════════════════════
# Products
# ═══════════════════════════════════════════════════════════


class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    price: float = Field(gt=0)
    stock: int = Field(ge=0)
    category_id: int


class ProductUpdate(BaseModel):
    name: str | None = None
    price: float | None = Field(default=None, gt=0)
    stock: int | None = Field(default=None, ge=0)


class Product(BaseModel):
    id: int
    name: str
    price: float
    stock: int
    category_id: int
    created_at: str


class ProductSummary(BaseModel):
    """For lists (fewer fields to reduce the payload)."""
    id: int
    name: str
    price: float


# ═══════════════════════════════════════════════════════════
# Users
# ═══════════════════════════════════════════════════════════


class UserUpdate(BaseModel):
    name: str | None = None
    email: str | None = None
    bio: str | None = Field(default=None, max_length=500)


class User(BaseModel):
    id: int
    name: str
    email: str
    bio: str = ""
    tier: Tier
    created_at: str


# ═══════════════════════════════════════════════════════════
# Categories
# ═══════════════════════════════════════════════════════════


class Category(BaseModel):
    id: int
    name: str
    parent_id: int | None = None


# ═══════════════════════════════════════════════════════════
# Orders
# ═══════════════════════════════════════════════════════════


class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(gt=0)


class OrderCreate(BaseModel):
    items: list[OrderItem]


class Order(BaseModel):
    id: int
    user_id: int
    items: list[OrderItem]
    total: float
    status: OrderStatus
    created_at: str


# ═══════════════════════════════════════════════════════════
# Analytics
# ═══════════════════════════════════════════════════════════


class AnalyticsEvent(BaseModel):
    event_type: str
    properties: dict = Field(default_factory=dict)


# ═══════════════════════════════════════════════════════════
# Common responses
# ═══════════════════════════════════════════════════════════


class HealthResponse(BaseModel):
    status: str  # "healthy" | "degraded"
    services: dict
    message: str | None = None


class MetricsResponse(BaseModel):
    cache_hits: int
    cache_misses: int
    hit_rate_percent: float
    rate_limited_requests: int
    active_sessions: int
    ws_clients_connected: int


class RateLimitErrorResponse(BaseModel):
    error: str = "rate_limit_exceeded"
    message: str
    retry_after_seconds: int
    category: str

The mock database (we simulate PostgreSQL)

Create app/db.py:

"""
A mock DB. In production this would be PostgreSQL + SQLAlchemy.
"""
import asyncio
from datetime import datetime


# We simulate PostgreSQL's latency (50 ms)
DB_LATENCY = 0.05


# In-memory data
products_db: dict[int, dict] = {}
categories_db: dict[int, dict] = {}
users_db: dict[int, dict] = {}
orders_db: dict[int, dict] = {}
analytics_events: list[dict] = []


def seed_data():
    """Load the initial data for development."""
    # Categories
    categories_db.update({
        1: {"id": 1, "name": "Electronics", "parent_id": None},
        2: {"id": 2, "name": "Books", "parent_id": None},
        3: {"id": 3, "name": "Laptops", "parent_id": 1},
        4: {"id": 4, "name": "Phones", "parent_id": 1},
    })

    # Products
    for i in range(1, 51):
        products_db[i] = {
            "id": i,
            "name": f"Product {i}",
            "price": 100.0 + (i * 7) % 500,
            "stock": (i * 13) % 100,
            "category_id": (i % 4) + 1,
            "created_at": datetime.now().isoformat(),
        }

    # Users with tiers
    users_db.update({
        1: {"id": 1, "name": "Alice", "email": "alice@x.com", "bio": "", "tier": "free", "password": "secret",
            "created_at": datetime.now().isoformat()},
        2: {"id": 2, "name": "Bob Pro", "email": "bob@x.com", "bio": "", "tier": "pro", "password": "secret",
            "created_at": datetime.now().isoformat()},
        3: {"id": 3, "name": "Carol Enterprise", "email": "carol@x.com", "bio": "", "tier": "enterprise", "password": "secret",
            "created_at": datetime.now().isoformat()},
    })


async def db_get_products(skip: int = 0, limit: int = 20) -> list[dict]:
    await asyncio.sleep(DB_LATENCY)
    items = list(products_db.values())[skip:skip + limit]
    return items


async def db_get_product(product_id: int) -> dict | None:
    await asyncio.sleep(DB_LATENCY)
    return products_db.get(product_id)


async def db_create_product(data: dict) -> dict:
    await asyncio.sleep(DB_LATENCY)
    new_id = max(products_db.keys(), default=0) + 1
    new_product = {
        "id": new_id,
        **data,
        "created_at": datetime.now().isoformat(),
    }
    products_db[new_id] = new_product
    return new_product


async def db_update_product(product_id: int, data: dict) -> dict | None:
    await asyncio.sleep(DB_LATENCY)
    if product_id not in products_db:
        return None
    products_db[product_id].update(data)
    return products_db[product_id]


async def db_delete_product(product_id: int) -> bool:
    await asyncio.sleep(DB_LATENCY)
    if product_id not in products_db:
        return False
    del products_db[product_id]
    return True


async def db_get_categories() -> list[dict]:
    await asyncio.sleep(DB_LATENCY)
    return list(categories_db.values())


async def db_get_user(user_id: int) -> dict | None:
    await asyncio.sleep(DB_LATENCY)
    user = users_db.get(user_id)
    if user:
        # Don't return the password
        return {k: v for k, v in user.items() if k != "password"}
    return None


async def db_update_user(user_id: int, data: dict) -> dict | None:
    await asyncio.sleep(DB_LATENCY)
    if user_id not in users_db:
        return None
    users_db[user_id].update(data)
    return await db_get_user(user_id)


async def db_authenticate_user(username: str, password: str) -> dict | None:
    await asyncio.sleep(DB_LATENCY)
    for u in users_db.values():
        if u["name"].lower().replace(" ", "") == username.lower() and u["password"] == password:
            return u
    return None


async def db_create_order(user_id: int, items: list[dict]) -> dict:
    await asyncio.sleep(DB_LATENCY * 2)  # slower (it's a transaction)

    # Validate the stock + calculate the total
    total = 0.0
    for item in items:
        product = products_db.get(item["product_id"])
        if not product:
            raise ValueError(f"Product {item['product_id']} not found")
        if product["stock"] < item["quantity"]:
            raise ValueError(f"Insufficient stock for product {item['product_id']}")
        total += product["price"] * item["quantity"]
        product["stock"] -= item["quantity"]

    # Create the order
    new_id = max(orders_db.keys(), default=0) + 1
    order = {
        "id": new_id,
        "user_id": user_id,
        "items": items,
        "total": total,
        "status": "pending",
        "created_at": datetime.now().isoformat(),
    }
    orders_db[new_id] = order
    return order

app/main.py (the scaffolding)

"""
Production Cached API - the FastAPI scaffolding.

This version is the SCAFFOLDING. Capsules 03-04 add the routers.
"""
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


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


# ═══════════════════════════════════════════════════════════
# Lifespan
# ═══════════════════════════════════════════════════════════


@asynccontextmanager
async def lifespan(app: FastAPI):
    # === Startup ===
    logger.info("Starting Production Cached API...")

    # 1. Init the Redis pool
    init_pool()

    # 2. Verify the Redis connection
    r = get_redis()
    try:
        await r.ping()
        logger.info("✓ Redis connected")
    except Exception as e:
        logger.error(f"✗ Redis connection failed: {e}")
        raise

    # 3. Seed the mock DB
    seed_data()
    logger.info("✓ Mock DB seeded")

    # 4. Start the background tasks (capsules 03-04 add them)
    # - The Pub/Sub listener
    # - The write-behind flusher
    # - The metrics reporter

    logger.info("✓ API ready")
    yield

    # === Shutdown ===
    logger.info("Shutting down...")
    await close_pool()
    logger.info("✓ Cleanup complete")


# ═══════════════════════════════════════════════════════════
# App
# ═══════════════════════════════════════════════════════════


app = FastAPI(
    title="Production Cached API",
    description="""
## An e-commerce API with advanced Redis

Features:
- A complete **caching strategy** (cache-aside, write-through, write-behind)
- Multi-tier sliding window **rate limiting**
- **JWT + Redis sessions** with global logout
- **Pub/Sub** for invalidation events
- **WebSockets** for real-time notifications
- **Graceful degradation** if Redis fails
    """,
    version="1.0.0",
    lifespan=lifespan,
)


# ═══════════════════════════════════════════════════════════
# Middleware (in execution order)
# ═══════════════════════════════════════════════════════════


# CORS (capsule 03 keeps this)
app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Capsules 03-04 add:
# - RateLimitMiddleware
# - CacheMiddleware
# - LoggingMiddleware


# ═══════════════════════════════════════════════════════════
# The health check (always available)
# ═══════════════════════════════════════════════════════════


@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"),
            "uptime_seconds": int(info.get("uptime_in_seconds", 0)),
            "memory_used": memory.get("used_memory_human"),
        }
    except Exception as e:
        health_data["status"] = "degraded"
        health_data["services"]["redis"] = {
            "status": "unhealthy",
            "reason": str(e),
        }
        health_data["message"] = "Redis unavailable. App in degraded mode."

    return health_data


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


# Capsules 03-04 add:
# - app.include_router(auth_router)
# - app.include_router(api_router)
# - app.include_router(analytics_router)
# - app.include_router(admin_router)
# - app.include_router(system_router)
# - The WebSocket endpoint

Verifying step 1

cd ~/projects/redis-guide/module-05-final/production-cached-api
source .venv/bin/activate

# Check the structure
ls app/
# config.py  redis_client.py  models.py  db.py  main.py  ...

# Check the imports
python -c "from app.models import Tier, Product, User; print('Models OK')"
python -c "from app.db import seed_data; seed_data(); from app.db import products_db; print(f'Products seeded: {len(products_db)}')"
python -c "from app.redis_client import init_pool, get_redis; init_pool(); print('Redis client OK')"

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

Expected output:

INFO:     Uvicorn running on http://127.0.0.1:8000
2026-04-25 [__main__] INFO: Starting Production Cached API...
2026-04-25 [app.redis_client] INFO: Redis pool initialized: max_connections=100
2026-04-25 [__main__] INFO: ✓ Redis connected
2026-04-25 [__main__] INFO: ✓ Mock DB seeded
2026-04-25 [__main__] INFO: ✓ API ready
INFO:     Application startup complete.

# Test the health check
curl http://localhost:8000/health | python -m json.tool

# Output:
# {
#   "status": "healthy",
#   "services": {
#     "redis": {
#       "status": "healthy",
#       "version": "7.2.4",
#       "uptime_seconds": 123,
#       "memory_used": "1.5M"
#     }
#   }
# }

✅ The scaffolding is complete. Capsules 03-04 can now add the routers.

Troubleshooting

Problem 1: An ImportError in the project's modules

Cause: The folder structure is missing the necessary __init__.py files.

Solution:

touch app/__init__.py
touch app/auth/__init__.py
touch app/caching/__init__.py
touch app/rate_limit/__init__.py
# ... etc

Problem 2: The lifespan doesn't run

Cause: You used @app.on_event("startup") (deprecated) instead of the lifespan.

Solution: Use the @asynccontextmanager with app = FastAPI(lifespan=lifespan).

Problem 3: RuntimeError: Redis pool not initialized

Cause: An endpoint called get_redis() but the lifespan never initialized the pool.

Solution: Make sure init_pool() is in the lifespan's startup, before the yield.

Problem 4: The configuration isn't read from the environment variables

Cause: Your venv or shell doesn't have the env vars exported.

Solution:

# Create .env
cat > .env << 'EOF'
REDIS_URL=redis://localhost:6379/0
JWT_SECRET=local-dev-secret
LOG_LEVEL=DEBUG
EOF

# Load it before uvicorn
export $(cat .env | xargs)
uvicorn app.main:app --reload

Or use python-dotenv:

pip install python-dotenv

# config.py
from dotenv import load_dotenv
load_dotenv()

Problem 5: The mock DB doesn't persist between restarts

Cause: That's deliberate — seed_data() runs on every startup. In development it's fine.

Solution: For a PR demo, use a local SQLite:

# Replace the mock dict with SQLAlchemy + SQLite
# (out of scope for this module, but the pattern is clear)

Exercises

Exercise 1: The complete setup (Easy)

Implement the project's full structure: the folders, __init__.py, config.py, redis_client.py, models.py, db.py, main.py. Verify that uvicorn app.main:app starts without errors and /health returns healthy.

mkdir -p production-cached-api && cd production-cached-api
mkdir -p app/auth app/caching app/rate_limit app/pubsub app/routers app/metrics tests scripts

touch app/__init__.py
for sub in auth caching rate_limit pubsub routers metrics; do
    touch app/$sub/__init__.py
done
touch tests/__init__.py

python -m venv .venv
source .venv/bin/activate
pip install fastapi "uvicorn[standard]" "redis>=7.4" pyjwt pydantic httpx websockets pytest pytest-asyncio

# Copy config.py, redis_client.py, models.py, db.py, main.py from the capsule

# Make sure Redis is up
docker ps | grep redis

uvicorn app.main:app --reload

curl http://localhost:8000/health
# {"status":"healthy", ...}

Exercise 2: The caching strategy document (Medium)

Write CACHING-STRATEGY.md with the complete endpoints + patterns matrix. If your app grows (you add more endpoints), you should be able to add them here.

Exercise 3: A mock DB with atomic consistency (Medium)

Modify db_create_order() so it uses an asyncio.Lock() to simulate transaction atomicity. Test with 2 concurrent orders for the same product (with stock=1) — only one should succeed.

import asyncio


_db_lock = asyncio.Lock()


async def db_create_order_atomic(user_id: int, items: list[dict]) -> dict:
    async with _db_lock:  # only one transaction at a time
        await asyncio.sleep(DB_LATENCY)

        total = 0.0
        for item in items:
            product = products_db.get(item["product_id"])
            if not product:
                raise ValueError(f"Product {item['product_id']} not found")
            if product["stock"] < item["quantity"]:
                raise ValueError(f"Insufficient stock for product {item['product_id']}")
            total += product["price"] * item["quantity"]
            product["stock"] -= item["quantity"]

        new_id = max(orders_db.keys(), default=0) + 1
        order = {
            "id": new_id,
            "user_id": user_id,
            "items": items,
            "total": total,
            "status": "pending",
            "created_at": datetime.now().isoformat(),
        }
        orders_db[new_id] = order
        return order

import asyncio


async def test_concurrent_orders():
    products_db[42] = {"id": 42, "name": "Limited", "price": 100.0, "stock": 1, "category_id": 1, "created_at": ""}

    async def order_attempt(user_id):
        try:
            return await db_create_order_atomic(user_id, [{"product_id": 42, "quantity": 1}])
        except ValueError as e:
            return f"failed: {e}"

    # 2 concurrent attempts
    results = await asyncio.gather(order_attempt(1), order_attempt(2))
    print(results)
    # Expected: one succeeds, one fails with "Insufficient stock"


asyncio.run(test_concurrent_orders())

Exercise 4: An initial /metrics endpoint (Easy-Medium)

Implement a GET /metrics endpoint that returns Redis's current usage (hits, misses, memory, connected clients). You'll extend it in the following capsules.

@app.get("/metrics")
async def metrics(r: Redis = Depends(get_redis_dep)):
    """The system's initial metrics."""
    try:
        info_stats = await r.info("stats")
        info_memory = await r.info("memory")
        info_clients = await r.info("clients")

        hits = info_stats.get("keyspace_hits", 0)
        misses = info_stats.get("keyspace_misses", 0)
        total = hits + misses
        hit_rate = (hits / total * 100) if total > 0 else 0

        return {
            "redis": {
                "version": (await r.info("server")).get("redis_version"),
                "memory_used": info_memory.get("used_memory_human"),
                "connected_clients": info_clients.get("connected_clients"),
                "total_commands_processed": info_stats.get("total_commands_processed"),
            },
            "cache": {
                "hits": hits,
                "misses": misses,
                "hit_rate_percent": round(hit_rate, 2),
            },
            # Capsules 03-04 add:
            # "rate_limiting": {...},
            # "sessions": {...},
            # "pubsub": {...},
        }
    except Exception as e:
        return {"error": str(e), "status": "degraded"}

# After making several requests
curl http://localhost:8000/metrics | python -m json.tool
# {
#   "redis": {"version": "7.2.4", "memory_used": "1.5M", ...},
#   "cache": {"hits": 12, "misses": 3, "hit_rate_percent": 80.0}
# }

Summary

In this capsule you built:

CACHING-STRATEGY.md — the document that defines which pattern each endpoint uses. It's a portfolio deliverable.

app/config.py — centralized configuration with tier limits, TTLs, endpoint mappings.

app/redis_client.py — the pool's singleton with init_pool(), get_redis(), close_pool().

app/models.py — Pydantic models for every endpoint (auth, products, users, orders, analytics).

app/db.py — a mock DB simulating PostgreSQL with artificial latency.

app/main.py — the FastAPI scaffolding with a lifespan event that initializes the Redis pool, verifies the connection, seeds the mock DB, and adds CORS middleware.

A health check that reports Redis's detailed status (version, memory, uptime).

The project's structure:

app/{auth,caching,rate_limit,pubsub,routers,metrics}/
tests/
scripts/
docker-compose.yml (capsule 05)
README.md (capsule 05)

The critical part: this is the professional scaffolding on top of which M5's capsules 03 and 04 build. The structure is clean, separated by concern, and ready to grow. The difference between a project that grows without chaos and one that turns into spaghetti is exactly this initial organization.

Additional resources

1. The Twelve-Factor App: Codebase — Principles for organizing projects
2. FastAPI Project Structure — The official recommendations
3. Pydantic v2 Models — Modern validation
4. Caching Strategy Documents — How to document caching strategies
5. Production-ready Python — A talk about project structure
6. 12-Factor: Config — Environment variables

What's next?

In Capsule 03 you fill the scaffolding with the caching + rate limiting routers: you implement the RateLimitMiddleware with a multi-tier sliding window, the CacheMiddleware that caches GET responses automatically, and the /api/products, /api/categories, and /api/users routers with working cache-aside. It's where the project comes to life.

Keep Redis running. Let's go.