Module 5: Capstone Project — API with Caching Strategy

Capsule 05 — End-to-end verification + Docker Compose + README + Portfolio

Overview

This is the closing capsule. Your Cached API works locally; here you leave it shipping-ready: anyone clones the repo, runs docker-compose up, and everything comes up with FastAPI + Postgres + Redis ready to go. You test the system end-to-end with a reproducible verify.sh script, write tests with pytest-asyncio, draft a professional README for your portfolio, and close the guide with a bridge to the next one (PostgreSQL & SQLAlchemy #8).

The tone: you're closing the guide, not teaching new features. Reuse what you already built. This capsule's goal is demonstrability: that you can show the project in an interview, that a colleague can bring it up in 2 minutes, and that you yourself understand it in 6 months when you open it again.

The project's final structure

After capsules 01-04, your tree should look like this:

cached-api/
├── app/
│   ├── __init__.py
│   ├── main.py                    # The FastAPI app, lifespan, /metrics, /healthz
│   ├── config.py                  # Settings (pydantic-settings)
│   ├── db.py                      # The SQLAlchemy async engine + session_maker
│   ├── models.py                  # Pydantic schemas + ORM models
│   ├── redis_client.py            # The ConnectionPool singleton
│   ├── dependencies.py            # get_redis, get_db
│   ├── metrics.py                 # Global counters
│   │
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── jwt_handler.py         # create/decode the JWT
│   │   ├── sessions.py            # Session CRUD in Redis
│   │   └── dependencies.py        # get_current_user
│   │
│   ├── cache/
│   │   ├── __init__.py
│   │   └── middleware.py          # The CacheMiddleware (auto-caches GETs)
│   │
│   ├── rate_limit/
│   │   ├── __init__.py
│   │   ├── algorithms.py          # check_sliding_window
│   │   ├── middleware.py          # The RateLimitMiddleware
│   │   └── tiers.py               # The TIER_LIMITS dict
│   │
│   ├── pubsub/
│   │   ├── __init__.py
│   │   └── listener.py            # The background cache:invalidate listener
│   │
│   ├── workers/
│   │   ├── __init__.py
│   │   └── analytics_worker.py    # The write-behind drainer
│   │
│   └── routers/
│       ├── __init__.py
│       ├── auth.py                # /auth/login, logout, sessions
│       ├── products.py            # /api/products (CRUD + cache)
│       ├── categories.py          # /api/categories
│       ├── users.py               # /api/users (write-through)
│       ├── search.py              # /api/search
│       ├── analytics.py           # /analytics/event (write-behind)
│       └── websocket.py           # /ws/notifications
│
├── tests/
│   ├── __init__.py
│   ├── conftest.py                # Fixtures: redis, db, client
│   ├── test_health.py
│   ├── test_auth.py
│   ├── test_cache.py
│   ├── test_rate_limit.py
│   └── test_analytics.py
│
├── scripts/
│   ├── verify.sh                  # An end-to-end smoke test
│   └── seed_db.py                 # Loads the initial products/categories
│
├── alembic/                       # DB migrations
│   ├── versions/
│   ├── env.py
│   └── script.py.mako
│
├── docker-compose.yml             # FastAPI + Postgres + Redis
├── Dockerfile                     # The API's image
├── requirements.txt
├── .env.example
├── .gitignore
├── alembic.ini
├── CACHING-STRATEGY.md            # The document you wrote in capsule 02
└── README.md

Note: if your folder has extra files (like notebooks/, dev_logs/), that's fine — but don't commit them to the public repo without cleaning them up first.

requirements.txt

fastapi>=0.136
uvicorn[standard]>=0.32
redis>=7.4
pydantic>=2.9
pydantic-settings>=2.6
sqlalchemy[asyncio]>=2.0.36
asyncpg>=0.30
alembic>=1.13.3
PyJWT>=2.12
pwdlib[bcrypt]>=0.3
python-multipart>=0.0.12

# Dev/test
pytest>=8.3
pytest-asyncio>=0.24
httpx>=0.27

A verified 2026 stack: redis>=7.4 with the redis.asyncio module (not the deprecated aioredis library). PyJWT (not python-jose, abandoned with unresolved CVEs in ecdsa). pwdlib (not passlib, broken with bcrypt 5.0).

Dockerfile

FROM python:3.13-slim

WORKDIR /app

# System deps for asyncpg
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libpq-dev curl \
    && rm -rf /var/lib/apt/lists/*

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

COPY . .

# Healthcheck
HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \
    CMD curl -fsS http://localhost:8000/healthz || exit 1

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

docker-compose.yml

The complete stack for local development. In production you'd swap the volumes for managed services.

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@postgres:5432/cached_api
      REDIS_URL: redis://redis:6379/0
      JWT_SECRET: ${JWT_SECRET:-dev-secret-change-in-prod}
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
    volumes:
      - ./app:/app/app  # Hot reload in dev
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

  postgres:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: cached_api
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    volumes:
      - pg_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 3s
      retries: 5
    ports:
      - "5432:5432"

  redis:
    image: redis:7.4-alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru --appendonly yes
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5
    ports:
      - "6379:6379"

volumes:
  pg_data:
  redis_data:

Why maxmemory-policy allkeys-lru: for a cache, you want Redis to evict automatically when memory fills up. allkeys-lru evicts the least-used ones. If you store critical data (sessions), consider volatile-lru (it only evicts keys with a TTL).

.env.example

# Copy to .env and adjust
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/cached_api
REDIS_URL=redis://localhost:6379/0
JWT_SECRET=change-me-with-secrets-token-urlsafe-32
JWT_EXPIRES_SECONDS=3600
SESSION_TTL_SECONDS=86400
CACHE_TTL_DEFAULT=60
CACHE_TTL_USER=300
CACHE_TTL_PRODUCT=120
CACHE_TTL_CATEGORY=3600
LOG_LEVEL=INFO

Generate a real secret:

python -c "import secrets; print(secrets.token_urlsafe(32))"

Local setup

Option A: Docker Compose (recommended)

git clone <your-repo>
cd cached-api
cp .env.example .env

# Bring up the stack
docker-compose up -d

# Wait for them to be healthy
docker-compose ps
# Every service should show "healthy"

# Migrations + seed
docker-compose exec api alembic upgrade head
docker-compose exec api python scripts/seed_db.py

# Verify
curl http://localhost:8000/healthz
# {"status": "ok", "redis": "ok", "db": "ok"}

Option B: External services + a venv

# This assumes Postgres and Redis are running locally
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env
alembic upgrade head
python scripts/seed_db.py

uvicorn app.main:app --reload

End-to-end verification: scripts/verify.sh

A reproducible script that tests the whole flow. Run it after bringing up the stack to confirm every feature works.

#!/usr/bin/env bash
# scripts/verify.sh
# An end-to-end smoke test. It assumes the API is at localhost:8000.

set -euo pipefail

BASE_URL="${BASE_URL:-http://localhost:8000}"
PASS=0
FAIL=0

run() {
    local name="$1"
    shift
    if "$@"; then
        echo "✓ $name"
        PASS=$((PASS+1))
    else
        echo "✗ $name"
        FAIL=$((FAIL+1))
    fi
}

check_status() {
    local expected="$1"
    local url="$2"
    shift 2
    local actual
    actual=$(curl -s -o /dev/null -w "%{http_code}" "$@" "$url")
    [[ "$actual" == "$expected" ]]
}

check_header() {
    local header_name="$1"
    local expected_value="$2"
    local url="$3"
    shift 3
    local actual
    actual=$(curl -s -D - -o /dev/null "$@" "$url" | grep -i "^$header_name:" | awk -F': ' '{print $2}' | tr -d '\r\n')
    [[ "$actual" == *"$expected_value"* ]]
}

echo "=== Cached API — End-to-end verification ==="
echo "Base URL: $BASE_URL"
echo ""

# 1. Health
echo "[1/8] Health checks"
run "GET /healthz returns 200" check_status 200 "$BASE_URL/healthz"
run "GET /metrics returns 200" check_status 200 "$BASE_URL/metrics"

# 2. Auth
echo ""
echo "[2/8] Authentication"
TOKEN=$(curl -s -X POST "$BASE_URL/auth/login" \
    -H "Content-Type: application/json" \
    -d '{"user_id": "alice", "tier": "free"}' | jq -r .access_token)

run "Login returns a token" test -n "$TOKEN"
run "Sessions list when authenticated" check_status 200 "$BASE_URL/auth/sessions" \
    -H "Authorization: Bearer $TOKEN"
run "Sessions list without auth returns 403" check_status 403 "$BASE_URL/auth/sessions"

# 3. Caching
echo ""
echo "[3/8] Caching"

# The first GET: MISS
check_header "x-cache" "MISS" "$BASE_URL/api/products" || true
# The second GET: HIT
run "The second GET /api/products is a HIT" \
    check_header "x-cache" "HIT" "$BASE_URL/api/products"

# Cache-Control: no-cache → BYPASS
run "Cache-Control: no-cache → BYPASS" \
    check_header "x-cache" "BYPASS" "$BASE_URL/api/products" \
    -H "Cache-Control: no-cache"

# 4. Rate limiting
echo ""
echo "[4/8] Rate limiting"

# Make 20 quick requests with no token (the free tier, e.g., 10/min)
RATE_LIMITED=false
for i in {1..20}; do
    code=$(curl -s -o /dev/null -w "%{http_code}" "$BASE_URL/api/products")
    if [[ "$code" == "429" ]]; then
        RATE_LIMITED=true
        break
    fi
done
run "The rate limit fires a 429" test "$RATE_LIMITED" = "true"
run "The X-RateLimit-Limit header is present" \
    check_header "x-ratelimit-limit" "" "$BASE_URL/api/products"

sleep 60  # Wait for the reset (in real life, configure a short window for testing)

# 5. Write-through
echo ""
echo "[5/8] Write-through (PUT /api/users/{id})"

# Reset the token (it may have expired)
TOKEN=$(curl -s -X POST "$BASE_URL/auth/login" \
    -H "Content-Type: application/json" \
    -d '{"user_id": "alice", "tier": "enterprise"}' | jq -r .access_token)

# The PUT updates
run "PUT /api/users/alice OK" check_status 200 "$BASE_URL/api/users/alice" \
    -X PUT -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"name": "Alice Test"}'

# The cache updated: the GET returns the new data
NAME=$(curl -s "$BASE_URL/api/users/alice" -H "Authorization: Bearer $TOKEN" | jq -r .name)
run "The cache shows the updated data" test "$NAME" = "Alice Test"

# 6. Write-behind
echo ""
echo "[6/8] Write-behind (POST /analytics/event)"

for i in {1..30}; do
    curl -s -X POST "$BASE_URL/analytics/event" \
        -H "Content-Type: application/json" \
        -d '{"event_type": "page_view", "user_id": "alice"}' > /dev/null
done

run "Events accepted (202)" \
    check_status 202 "$BASE_URL/analytics/event" \
    -X POST -H "Content-Type: application/json" \
    -d '{"event_type": "click"}'

echo "  Waiting for the worker's flush (6s)..."
sleep 6
QUEUE=$(redis-cli -u "${REDIS_URL:-redis://localhost:6379/0}" LLEN analytics:queue)
run "The queue was drained by the worker" test "$QUEUE" -lt 5

# 7. Sessions
echo ""
echo "[7/8] Sessions"

run "Logout returns 204" check_status 204 "$BASE_URL/auth/logout" \
    -X POST -H "Authorization: Bearer $TOKEN"

# The revoked token: an immediate 401
run "A revoked token: an immediate 401" check_status 401 "$BASE_URL/auth/sessions" \
    -H "Authorization: Bearer $TOKEN"

# 8. The final metrics
echo ""
echo "[8/8] Metrics"
METRICS=$(curl -s "$BASE_URL/metrics")
run "Metrics has cache.hits" test "$(echo "$METRICS" | jq '.cache.hits')" -gt 0
run "Metrics has rate_limit.blocked" test "$(echo "$METRICS" | jq '.rate_limit.blocked')" -gt 0
run "Metrics has sessions.active" test "$(echo "$METRICS" | jq '.sessions.active')" -ge 0

echo ""
echo "=== Results ==="
echo "PASS: $PASS"
echo "FAIL: $FAIL"

if [[ $FAIL -gt 0 ]]; then
    exit 1
fi

Run it:

chmod +x scripts/verify.sh
./scripts/verify.sh

Expected output: every check at ✓ and FAIL: 0. If something fails, read the check's name and go look at the corresponding endpoint.

Tests with pytest-asyncio

Automated tests that exercise the critical components. You don't need 100% coverage — focus on the flows you use most.

tests/conftest.py

# tests/conftest.py
import asyncio
import os
from typing import AsyncGenerator

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from redis.asyncio import Redis

# Set the test environment BEFORE importing the app
os.environ.setdefault("REDIS_URL", "redis://localhost:6379/15")  # DB 15 for tests
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://postgres:postgres@localhost:5432/cached_api_test")

from app.main import app
from app.redis_client import init_pool, get_pool, close_pool


@pytest.fixture(scope="session")
def event_loop():
    """A shared loop for the whole test session."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest_asyncio.fixture(scope="session", autouse=True)
async def setup_redis():
    """Initializes the pool and cleans the test DB at the start."""
    init_pool(os.environ["REDIS_URL"], max_connections=10)
    pool = get_pool()
    r = Redis(connection_pool=pool)
    await r.flushdb()  # Clean DB 15
    yield
    await r.flushdb()
    await close_pool()


@pytest_asyncio.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
    """An async HTTP client for hitting the app without a real server."""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac


@pytest_asyncio.fixture
async def redis_client() -> AsyncGenerator[Redis, None]:
    """A Redis client for direct checks."""
    pool = get_pool()
    r = Redis(connection_pool=pool)
    yield r


@pytest_asyncio.fixture
async def auth_token(client: AsyncClient) -> str:
    """A valid token for alice (the free tier)."""
    response = await client.post(
        "/auth/login",
        json={"user_id": "alice", "tier": "free"},
    )
    return response.json()["access_token"]

tests/test_health.py

# tests/test_health.py
import pytest


@pytest.mark.asyncio
async def test_healthz_returns_ok(client):
    response = await client.get("/healthz")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "ok"
    assert data.get("redis") == "ok"


@pytest.mark.asyncio
async def test_metrics_returns_dict(client):
    response = await client.get("/metrics")
    assert response.status_code == 200
    data = response.json()
    assert "cache" in data
    assert "rate_limit" in data
    assert "sessions" in data

tests/test_auth.py

# tests/test_auth.py
import pytest


@pytest.mark.asyncio
async def test_login_returns_token(client):
    response = await client.post(
        "/auth/login",
        json={"user_id": "alice", "tier": "free"},
    )
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"


@pytest.mark.asyncio
async def test_login_unknown_user_returns_401(client):
    response = await client.post(
        "/auth/login",
        json={"user_id": "ghost", "tier": "free"},
    )
    assert response.status_code == 401


@pytest.mark.asyncio
async def test_sessions_requires_auth(client):
    response = await client.get("/auth/sessions")
    assert response.status_code == 403


@pytest.mark.asyncio
async def test_logout_revokes_session(client, auth_token):
    # Sessions OK with the token
    response = await client.get(
        "/auth/sessions",
        headers={"Authorization": f"Bearer {auth_token}"},
    )
    assert response.status_code == 200

    # Logout
    response = await client.post(
        "/auth/logout",
        headers={"Authorization": f"Bearer {auth_token}"},
    )
    assert response.status_code == 204

    # The same token: an immediate 401
    response = await client.get(
        "/auth/sessions",
        headers={"Authorization": f"Bearer {auth_token}"},
    )
    assert response.status_code == 401

tests/test_cache.py

# tests/test_cache.py
import pytest


@pytest.mark.asyncio
async def test_first_get_is_miss_second_is_hit(client, redis_client):
    # Clean the cache
    cursor = 0
    while True:
        cursor, batch = await redis_client.scan(cursor, match="cache:*", count=100)
        if batch:
            await redis_client.delete(*batch)
        if cursor == 0:
            break

    # The first GET: MISS
    r1 = await client.get("/api/products")
    assert r1.headers.get("x-cache", "").upper() == "MISS"

    # The second GET: HIT
    r2 = await client.get("/api/products")
    assert r2.headers.get("x-cache", "").upper() == "HIT"


@pytest.mark.asyncio
async def test_no_cache_header_bypasses(client):
    response = await client.get(
        "/api/products",
        headers={"Cache-Control": "no-cache"},
    )
    assert response.headers.get("x-cache", "").upper() == "BYPASS"

tests/test_rate_limit.py

# tests/test_rate_limit.py
import pytest


@pytest.mark.asyncio
async def test_rate_limit_eventually_blocks(client, redis_client):
    # Clean the rate keys
    cursor = 0
    while True:
        cursor, batch = await redis_client.scan(cursor, match="rate:*", count=100)
        if batch:
            await redis_client.delete(*batch)
        if cursor == 0:
            break

    blocked = False
    for _ in range(50):
        response = await client.get("/api/products")
        if response.status_code == 429:
            blocked = True
            assert "retry-after" in {k.lower() for k in response.headers.keys()}
            break

    assert blocked, "The rate limit never fired after 50 requests"

Running the tests

# Locally
pytest -v

# Docker Compose
docker-compose exec api pytest -v

A note about the tests: these are integrated smoke tests (hitting real endpoints against a real Redis). They're slow but realistic. If you want unit tests for pure functions (check_sliding_window, create_session), add them in tests/test_units.py. The Testing guide (#11) covers the full spectrum.

A professional README

A template for the root README.md. Copy it, adapt the names, add optional screenshots.

# Cached API — a production-ready FastAPI + Redis service

> The capstone project of the Backend Python Bootcamp — the Redis & Caching Strategies guide.

A REST API with multi-level caching, rate limiting per tier, distributed sessions, write-through/write-behind, Pub/Sub for invalidation, WebSockets for real-time notifications, and observability metrics. **It demonstrates command of Redis in production.**

---

## Stack

- **API**: FastAPI 0.115 + Uvicorn
- **Cache + Sessions + Pub/Sub**: Redis 7.4
- **DB**: PostgreSQL 17 (asyncpg)
- **Auth**: JWT (PyJWT) + Redis sessions
- **ORM**: SQLAlchemy 2.0 async
- **Migrations**: Alembic
- **Tests**: pytest + pytest-asyncio + httpx
- **Container**: Docker + docker-compose

---

## Features

| Feature | The pattern applied | Endpoint |
|---|---|---|
| Auth with JWT | JWT + Redis session HASHes | `/auth/login`, `/auth/logout`, `/auth/sessions` |
| A cached listing | Cache-aside with automatic middleware | `GET /api/products` |
| A cached detail | Cache-aside + a sliding TTL | `GET /api/products/{id}` |
| A consistent update | Write-through + Pub/Sub | `PUT /api/users/{id}` |
| Scalable analytics | Write-behind + an async worker | `POST /analytics/event` |
| Tiered rate limiting | A sliding window with Sorted Sets | All of `/api/*` |
| Real-time notifs | A Pub/Sub → WebSocket bridge | `WS /ws/notifications` |
| Observability | Counters + Redis info | `GET /metrics`, `/healthz` |

---

## Architecture

```
An HTTP client → [Middleware: RateLimit → Cache] → Router → DB / Redis
                       ↓                              ↑
                  Redis (rate keys)              Redis (cache + sessions)
                                                       ↓
                                                   Pub/Sub
                                                       ↓
                            ┌──────────────────────────┴───────────────────┐
                            ↓                                              ↓
                    The Background Listener                      The WebSocket Manager
                    (invalidates caches,                          (pushes notifs to
                     increments metrics)                           connected clients)
```

See `CACHING-STRATEGY.md` for the decision per endpoint (which pattern, which TTL, why).

---

## Setup

### With Docker Compose (recommended)

```bash
git clone <this-repo>
cd cached-api
cp .env.example .env

docker-compose up -d
docker-compose exec api alembic upgrade head
docker-compose exec api python scripts/seed_db.py

curl http://localhost:8000/healthz
```

The API is at http://localhost:8000. Interactive docs at http://localhost:8000/docs.

### Locally with a venv

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

# This assumes Postgres and Redis are running locally
cp .env.example .env
alembic upgrade head
python scripts/seed_db.py

uvicorn app.main:app --reload
```

---

## Verification

```bash
chmod +x scripts/verify.sh
./scripts/verify.sh
```

Output: every check ✓.

### Automated tests

```bash
pytest -v
```

---

## Live metrics

```bash
curl http://localhost:8000/metrics | jq
```

```json
{
  "cache": { "hits": 1245, "misses": 87, "hit_ratio": 0.93 },
  "rate_limit": { "blocked": 23 },
  "pubsub": { "events_received": 145, "by_type": { "user.updated": 12, "product.created": 5 } },
  "sessions": { "active": 8 },
  "analytics": { "queue_length": 0 }
}
```

---

## Relevant technical decisions

- **`redis.asyncio` instead of `aioredis`**: the `aioredis` library has been deprecated since 2022. The official module lives inside `redis-py` ≥ 4.2. A verified 2026 stack.
- **JWT + Redis sessions**: the JWT identifies the user, but the session lives in Redis (a HASH). It enables immediate logout (no waiting for the token to expire).
- **A sliding window with Sorted Sets**: precise and memory-efficient rate limiting. Chosen over a token bucket for simplicity and atomicity with a pipeline.
- **A ConnectionPool singleton**: one global pool shared via DI. It avoids recreating connections per request. Configured with `max_connections=50`.
- **Automatic CacheMiddleware**: whether a GET is cacheable is decided by the endpoint's `Cache-Control: max-age=`. The client can skip it with `Cache-Control: no-cache`.
- **Graceful degradation**: if Redis goes down, the middleware logs the error and lets the request through to the handler (it uses the DB directly). The cache isn't the source of truth.

---

## Repo structure

```
app/                # FastAPI application code
├── auth/           # JWT + sessions
├── cache/          # Caching middleware
├── rate_limit/     # Sliding window + tiers
├── pubsub/         # Background listener
├── workers/        # Write-behind drainer
└── routers/        # Endpoints

tests/              # pytest-asyncio
scripts/            # verify.sh, seed_db.py
alembic/            # DB migrations
docker-compose.yml
Dockerfile
CACHING-STRATEGY.md # Caching decisions per endpoint
```

---

## What I learned

This project is the capstone of the **Redis & Caching Strategies** guide from the Backend Python Bootcamp. It applies patterns from the modules:

1. **Redis Fundamentals** — Strings, Hashes, Lists, Sets, Sorted Sets
2. **Caching Patterns** — Cache-aside, write-through, write-behind, TTL strategies, stampede prevention
3. **Rate Limiting & Sessions** — Sliding window, JWT + sessions
4. **Pub/Sub & FastAPI Integration** — async pooling, lifespan events, middleware

Next step: replace module 1's "in-memory DB" with **real PostgreSQL and SQLAlchemy** (Guide #8 of the bootcamp).

---

## License

MIT — use it, modify it, share it.

A portfolio tip: add a GIF/screenshot of the flow (login → request → seeing the X-Cache: HIT headers). It takes 5 minutes and it raises the README a lot in a recruiter's eyes.

Portfolio: how to present this project

On GitHub

  1. A public repo with a clear README (the one above).
  2. Repo topics: fastapi, redis, caching, rate-limiting, python, docker, postgresql.
  3. The repo's About (the right panel): "Production-ready FastAPI service with Redis caching, rate limiting, sessions, and Pub/Sub. Capstone project."
  4. Pinned on your GitHub profile.
  5. An optional live demo: deploy it on Railway/Fly.io with a managed Redis. Paste the link in the README.

On your CV / LinkedIn

Cached API — Production FastAPI service (personal project, 2026) Built a high-throughput API demonstrating cache-aside, write-through, and write-behind patterns with Redis. Implemented sliding-window rate limiting (precise to the millisecond), JWT + Redis distributed sessions with immediate logout, Pub/Sub-based cache invalidation, and a WebSocket bridge for real-time notifications. Containerized with Docker Compose (FastAPI + PostgreSQL + Redis). Tested with pytest-asyncio. Stack: Python 3.13, FastAPI, Redis, PostgreSQL, SQLAlchemy async, JWT, Docker.

In technical interviews

When they ask you "tell me about a project you've built", steer the conversation to:

  1. The problem it solves: a modern API with multi-pattern caching and tiered rate limiting.
  2. One concrete technical decision: for example, "I chose a sliding window with Sorted Sets over a token bucket because..." or "for the analytics endpoint, I used write-behind because eventual durability is acceptable and I need maximum throughput."
  3. A trade-off you understand: "write-behind risks data loss if Redis goes down with a full queue, but for non-critical analytics that trade-off is worth the throughput."

If they ask you to show the code live, open CACHING-STRATEGY.md first (the per-endpoint decisions document). That shows you designed before you coded.

The connection with the next guide

This project uses Postgres, but with a minimal schema and seed data. Guide #8 — PostgreSQL & SQLAlchemy goes deeper:

  • Advanced relational modeling (joins, indexes, constraints)
  • Query optimization (EXPLAIN ANALYZE, the N+1 problem)
  • Complex migrations with Alembic
  • DB connection pooling (similar to Redis's)
  • Transactions, isolation levels
  • Read replicas, basic sharding

What changes: in this guide Redis is the protagonist and Postgres is the support. In guide 8 it's the other way around.

When you finish guide 8, come back to this repo and refactor the data model. You'll see how much your mental model grew.

The project's final checklist

Before you declare the project shipping-ready, verify:

  • docker-compose up brings up the API + Postgres + Redis with no errors
  • docker-compose ps shows all 3 services as healthy
  • alembic upgrade head runs with no errors
  • python scripts/seed_db.py loads the initial data
  • curl /healthz returns a 200 with redis: ok and db: ok
  • curl /docs opens the Swagger UI
  • Logging in with alice returns a valid token
  • A second call to GET /api/products returns X-Cache: HIT
  • Cache-Control: no-cache returns X-Cache: BYPASS
  • The rate limit fires a 429 on the free tier after N requests
  • The X-RateLimit-Limit, X-RateLimit-Remaining, and Retry-After headers are present
  • A PUT to a user updates Postgres + the cache (write-through)
  • A POST to analytics returns an immediate 202
  • The worker drains analytics:queue into Postgres in <10s
  • A logout with a valid token → the next request is a 401 (immediate logout)
  • The WebSocket connects with ?token= and receives events when the user is updated
  • /metrics returns a dict with cache, rate_limit, sessions, pubsub, analytics
  • scripts/verify.sh passes with no failures
  • pytest -v passes every test
  • CACHING-STRATEGY.md documents the decision per endpoint
  • The README has the setup, features, architecture, and technical decisions
  • .env.example has no real secrets (only placeholders)
  • .gitignore includes .env, .venv/, __pycache__/, *.pyc
  • Clean commits (no junk files, no accidental node_modules)
  • A public repo on GitHub with topics and a description
  • Pinned on your profile

If the checklist is 100%, the project is portfolio-ready. Share it.

Closing the guide

You made it to the end. A review of what you now command:

Module 1 — Redis Fundamentals

  • Strings, Hashes, Lists, Sets, Sorted Sets — when to use each one
  • Atomic INCR, TTL, EXPIRE
  • redis-py / redis.asyncio: the official clients

Module 2 — Caching Patterns

  • Cache-aside (lazy load), write-through (consistency), write-behind (throughput)
  • TTL strategies (fixed vs sliding), invalidation (manual / event-driven / version-based)
  • Cache stampede prevention with locks

Module 3 — Rate Limiting & Sessions

  • Token bucket, leaky bucket, sliding window — the comparison
  • A sliding window with Sorted Sets (the professional implementation)
  • JWT + Redis sessions: immediate logout, multi-device, a sliding TTL

Module 4 — Pub/Sub & FastAPI

  • PUBLISH/SUBSCRIBE/PSUBSCRIBE, fire-and-forget
  • A ConnectionPool singleton, lifespan events
  • The CacheMiddleware, graceful degradation
  • A WebSocket bridge

Module 5 — Production Cached API

  • The complete integration: everything above in a shipping-ready project
  • Docker Compose, tests, a README, a portfolio

You have the building blocks to solve caching, rate limiting, and messaging problems in any API you build. What comes in advanced bootcamps (PostgreSQL, microservices, observability) rests on these foundations.

Next stop: Guide #8 — PostgreSQL & SQLAlchemy.

Let's go.

Final resources

  1. Redis Best Practices — Optimization in production
  2. FastAPI in Production — Deployment patterns
  3. 12-Factor App — The manifesto for cloud-native apps
  4. The Twelve-Factor App: Backing services — How to treat Redis/Postgres as resources
  5. Awesome Redis — A curated list of resources
  6. SRE: Latency and Performance — If cache vs latency interests you
  7. Designing Data-Intensive Applications — The reference book for distributed systems (Kleppmann)
  8. High Performance Browser Networking — How HTTP/WebSockets interact with caches
← PreviousCapsule 04 — Sessions + Pub/Sub + Monitoring