Module 3: Rate Limiting and Session Storage

Mini-project: Rate Limiter Service

Next →

Overview

You close module 3 with an integrative project that combines everything you've learned: sliding window rate limiting with sorted sets, sessions complementing JWT, multi-tier limits per user plan, and concurrent load tests to verify the system behaves correctly under pressure.

The Rate Limiter Service is a complete FastAPI API anyone could add to their production project. It has 4 integrated features: professional rate limiting with a sliding window, authentication with JWT + Redis sessions (including local logout, global logout, and list devices), three pricing tiers (free/pro/enterprise) with differentiated limits, and a load test with asyncio + httpx that fires 500 concurrent requests to verify atomicity. It's portfolio-worthy: copy-paste it into your GitHub repo with a professional README and you have a project that demonstrates command of advanced Redis.

It isn't a step-by-step tutorial — it's clear specs and reference code. You can implement it from scratch (recommended) or copy the code and understand it. What matters is understanding why each architectural decision is where it is: why the rate limiting is middleware, why the sessions are DI, why the tests are async. You've seen every decision justified in earlier capsules; here you assemble them.

The project's specs

Endpoints

POST /auth/login                         → Login + create session + JWT
POST /auth/logout                        → Local logout
POST /auth/logout-all                    → Logout on every device
GET  /auth/sessions                      → List my active devices
POST /auth/sessions/{sid}/revoke         → Revoke a specific device

GET  /api/products                       → A normal endpoint (rate-limited by tier)
GET  /api/search?q=...                   → An expensive endpoint (a stricter rate limit)
POST /api/orders                         → A write endpoint (a moderate rate limit)

GET  /me                                 → Info about my current user/session
GET  /admin/users/{user_id}/revoke       → Admin: force a global logout for a user
GET  /health                             → Health check (not rate-limited)
GET  /metrics                            → Internal metrics (admin only)

Rate limits per tier

TierGeneralSearchOrders
free100/hr20/hr10/hr
pro1000/hr200/hr100/hr
enterprise10000/hr2000/hr1000/hr

Behavior

  • The JWT has a 24h TTL. A sliding TTL on the sessions.
  • A sliding window for rate limiting (not a fixed window).
  • Standard HTTP headers: X-RateLimit-*, Retry-After, 429 Too Many Requests.
  • Graceful degradation: if Redis goes down, the API allows requests with warning logs (no rate limit applied).
  • Structured logging: auth events (login, logout, revoke), rate limit hits.

Project structure

rate-limiter-service/
├── .venv/
├── app/
│   ├── __init__.py
│   ├── main.py              # The FastAPI app + middleware
│   ├── config.py            # Configuration (Redis URL, tiers, JWT)
│   ├── auth.py              # JWT + sessions
│   ├── rate_limiter.py      # Sliding window + multi-tier
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── auth.py          # /auth/* endpoints
│   │   ├── api.py           # /api/* endpoints (the protected ones)
│   │   └── admin.py         # /admin/* endpoints
│   └── models.py            # Pydantic models
├── tests/
│   ├── __init__.py
│   ├── test_auth.py
│   └── test_rate_limit_load.py
├── requirements.txt
├── .env.example
└── README.md

requirements.txt

fastapi>=0.136
uvicorn[standard]>=0.27.0
redis>=7.4
pyjwt>=2.8.0
pydantic>=2.0
httpx>=0.26.0
pytest>=8.0.0
pytest-asyncio>=0.23.0

app/config.py

"""
The Rate Limiter Service's configuration.
"""
import os


# Redis
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

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

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

# The rate limit per tier (general / search / orders)
TIER_LIMITS = {
    "free": {
        "general": (100, 3600),    # 100/hr
        "search":  (20, 3600),     # 20/hr
        "orders":  (10, 3600),     # 10/hr
    },
    "pro": {
        "general": (1000, 3600),
        "search":  (200, 3600),
        "orders":  (100, 3600),
    },
    "enterprise": {
        "general": (10000, 3600),
        "search":  (2000, 3600),
        "orders":  (1000, 3600),
    },
}

# Endpoint → rate limit category
ENDPOINT_CATEGORY = {
    "GET:/api/products": "general",
    "GET:/api/search": "search",
    "POST:/api/orders": "orders",
}
DEFAULT_CATEGORY = "general"

app/rate_limiter.py

"""
A sliding window rate limiter with multi-tier support.
"""
import time
import uuid
import redis
from redis.exceptions import RedisError
import logging

from app.config import REDIS_URL, TIER_LIMITS, ENDPOINT_CATEGORY, DEFAULT_CATEGORY


logger = logging.getLogger("rate_limiter")


r = redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=2)


def get_limit_for(method: str, path: str, tier: str) -> tuple[int, int]:
    """Returns: (limit, window_seconds) for the endpoint and tier."""
    endpoint_key = f"{method}:{path}"
    category = ENDPOINT_CATEGORY.get(endpoint_key, DEFAULT_CATEGORY)

    tier_config = TIER_LIMITS.get(tier, TIER_LIMITS["free"])
    return tier_config[category]


def check_rate_limit(
    user_id: str,
    method: str,
    path: str,
    tier: str = "free",
) -> 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, category
    """
    limit, window = get_limit_for(method, path, tier)
    category = ENDPOINT_CATEGORY.get(f"{method}:{path}", DEFAULT_CATEGORY)

    key = f"rate:{tier}:{category}:user:{user_id}"
    now = time.time()
    window_start = now - window
    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 + 60)
        _, _, count, _ = pipe.execute()

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

            # Calculate retry_after based on oldest entry
            oldest = r.zrange(key, 0, 0, withscores=True)
            if oldest:
                oldest_score = oldest[0][1]
                retry_after = max(0, oldest_score + window - now)
            else:
                retry_after = window

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

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

    except RedisError as e:
        # Graceful degradation: allow request when Redis is down
        logger.warning(f"Rate limit check failed (Redis error): {e}. Allowing request.")
        return (True, {
            "limit": limit,
            "remaining": limit,
            "retry_after_seconds": 0,
            "reset_in_seconds": int(window),
            "category": category,
            "degraded": True,
        })

app/auth.py

"""
JWT + Redis sessions.
"""
import time
import uuid
import secrets
import jwt
import redis
from redis.exceptions import RedisError
import logging

from app.config import (
    REDIS_URL, JWT_SECRET, JWT_ALGORITHM, JWT_TTL_HOURS,
    SESSION_TTL_SECONDS, SESSION_ABSOLUTE_TTL_SECONDS,
    MAX_SESSIONS_PER_USER,
)


logger = logging.getLogger("auth")


r = redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=2)


# ═══════════════════════════════════════════════════════════
# JWT
# ═══════════════════════════════════════════════════════════


def create_jwt(user_id: int, session_id: str) -> str:
    payload = {
        "user_id": user_id,
        "session_id": session_id,
        "iat": int(time.time()),
        "exp": int(time.time() + JWT_TTL_HOURS * 3600),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)


def decode_jwt(token: str) -> dict:
    return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])


# ═══════════════════════════════════════════════════════════
# Sessions
# ═══════════════════════════════════════════════════════════


def create_session(user_id: int, tier: str, metadata: dict | None = None) -> str:
    session_id = secrets.token_urlsafe(24)
    now = time.time()
    metadata = metadata or {}

    pipe = r.pipeline()
    pipe.hset(f"session:{session_id}", mapping={
        "user_id": str(user_id),
        "tier": tier,
        "created_at": str(now),
        "absolute_expires_at": str(now + SESSION_ABSOLUTE_TTL_SECONDS),
        "last_accessed": str(now),
        "ip": metadata.get("ip", ""),
        "user_agent": metadata.get("user_agent", "")[:200],
        "device_name": metadata.get("device_name", "Unknown"),
    })
    pipe.expire(f"session:{session_id}", SESSION_TTL_SECONDS)
    pipe.sadd(f"user:{user_id}:sessions", session_id)
    pipe.expire(f"user:{user_id}:sessions", SESSION_ABSOLUTE_TTL_SECONDS)
    pipe.execute()

    enforce_session_limit(user_id)

    logger.info(f"Session created: user_id={user_id}, sid={session_id[:16]}..., device={metadata.get('device_name')}")
    return session_id


def enforce_session_limit(user_id: int):
    sids = r.smembers(f"user:{user_id}:sessions")
    if len(sids) <= MAX_SESSIONS_PER_USER:
        return

    valid_sessions = []
    for sid in sids:
        data = r.hgetall(f"session:{sid}")
        if data:
            valid_sessions.append((sid, float(data["created_at"])))
        else:
            r.srem(f"user:{user_id}:sessions", sid)

    if len(valid_sessions) > MAX_SESSIONS_PER_USER:
        # Remove the oldest ones
        valid_sessions.sort(key=lambda x: x[1])
        to_remove = valid_sessions[:-MAX_SESSIONS_PER_USER]
        for sid, _ in to_remove:
            revoke_session(sid)
            logger.info(f"Session evicted (limit): user_id={user_id}, sid={sid[:16]}...")


def get_session(session_id: str) -> dict | None:
    try:
        data = r.hgetall(f"session:{session_id}")
        if not data:
            return None

        # Check absolute TTL
        absolute_expires = float(data.get("absolute_expires_at", 0))
        if time.time() > absolute_expires:
            r.delete(f"session:{session_id}")
            return None

        # Sliding TTL renewal
        pipe = r.pipeline()
        pipe.expire(f"session:{session_id}", SESSION_TTL_SECONDS)
        pipe.hset(f"session:{session_id}", "last_accessed", str(time.time()))
        pipe.execute()

        return data
    except RedisError as e:
        logger.error(f"Session read failed: {e}")
        return None


def revoke_session(session_id: str) -> bool:
    data = r.hgetall(f"session:{session_id}")
    if not data:
        return False

    user_id = data.get("user_id")
    pipe = r.pipeline()
    pipe.delete(f"session:{session_id}")
    if user_id:
        pipe.srem(f"user:{user_id}:sessions", session_id)
    pipe.execute()

    logger.info(f"Session revoked: sid={session_id[:16]}...")
    return True


def revoke_all_sessions(user_id: int) -> int:
    sids = r.smembers(f"user:{user_id}:sessions")
    if not sids:
        return 0

    pipe = r.pipeline()
    for sid in sids:
        pipe.delete(f"session:{sid}")
    pipe.delete(f"user:{user_id}:sessions")
    pipe.execute()

    logger.info(f"All sessions revoked: user_id={user_id}, count={len(sids)}")
    return len(sids)


def list_user_sessions(user_id: int) -> list[dict]:
    sids = r.smembers(f"user:{user_id}:sessions")
    sessions = []

    for sid in sids:
        data = r.hgetall(f"session:{sid}")
        if data:
            sessions.append({"session_id": sid, **data})
        else:
            r.srem(f"user:{user_id}:sessions", sid)

    return sessions

app/main.py (the main FastAPI app)

"""
Rate Limiter Service - the main FastAPI app.
"""
import time
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import jwt as jwt_lib

from app.auth import decode_jwt, get_session
from app.rate_limiter import check_rate_limit
from app.routers import auth as auth_router, api as api_router, admin as admin_router


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


# ═══════════════════════════════════════════════════════════
# The auth dependency (shared)
# ═══════════════════════════════════════════════════════════


security = HTTPBearer()


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    try:
        payload = decode_jwt(credentials.credentials)
    except jwt_lib.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt_lib.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

    session_id = payload.get("session_id")
    if not session_id:
        raise HTTPException(401, "Invalid token: no session_id")

    session = get_session(session_id)
    if not session:
        raise HTTPException(401, "Session revoked or expired. Please login again.")

    return {
        "user_id": int(session["user_id"]),
        "session_id": session_id,
        "tier": session.get("tier", "free"),
        "device_name": session.get("device_name", ""),
    }


# ═══════════════════════════════════════════════════════════
# The rate limit middleware
# ═══════════════════════════════════════════════════════════


SKIP_RATE_LIMIT_PATHS = {"/health", "/auth/login", "/docs", "/openapi.json"}


class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in SKIP_RATE_LIMIT_PATHS:
            return await call_next(request)

        # Try to extract user_id and tier from JWT
        user_id = None
        tier = "free"

        auth_header = request.headers.get("authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header.split(" ", 1)[1]
            try:
                payload = decode_jwt(token)
                session_id = payload.get("session_id")
                if session_id:
                    session = get_session(session_id)
                    if session:
                        user_id = session.get("user_id")
                        tier = session.get("tier", "free")
            except Exception:
                pass  # An invalid token — the endpoint will handle the 401

        # Rate limit per user (if authenticated) or per IP
        rate_key = f"user:{user_id}" if user_id else f"ip:{request.client.host if request.client else 'unknown'}"

        allowed, info = check_rate_limit(
            user_id=rate_key,
            method=request.method,
            path=request.url.path,
            tier=tier,
        )

        if not allowed:
            return JSONResponse(
                status_code=429,
                content={
                    "error": "rate_limit_exceeded",
                    "message": f"Rate limit exceeded for {info['category']}. Retry after {int(info['retry_after_seconds'])}s.",
                    "retry_after_seconds": int(info["retry_after_seconds"]),
                    "category": info["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"])),
                }
            )

        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


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


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Rate Limiter Service started")
    yield
    logger.info("Rate Limiter Service stopped")


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


app = FastAPI(
    title="Rate Limiter Service",
    description="An API with professional rate limiting + JWT + Redis sessions",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(RateLimitMiddleware)


# Routers
app.include_router(auth_router.router)
app.include_router(api_router.router)
app.include_router(admin_router.router)


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


@app.get("/me")
def me(user: dict = Depends(get_current_user)):
    return {
        "user_id": user["user_id"],
        "tier": user["tier"],
        "device": user["device_name"],
        "session_id": user["session_id"],
    }

app/routers/auth.py

"""
Auth endpoints: /auth/login, /auth/logout, /auth/sessions, etc.
"""
from fastapi import APIRouter, HTTPException, Request, Depends
from pydantic import BaseModel
from app.auth import (
    create_jwt,
    create_session,
    revoke_session,
    revoke_all_sessions,
    list_user_sessions,
)


router = APIRouter(prefix="/auth", tags=["Auth"])


# A mock user database (in production: PostgreSQL + bcrypt)
fake_users_db = {
    "alice":   {"user_id": 1, "username": "alice",   "password": "secret", "tier": "free"},
    "bob_pro": {"user_id": 2, "username": "bob_pro", "password": "secret", "tier": "pro"},
    "ent_carol": {"user_id": 3, "username": "ent_carol", "password": "secret", "tier": "enterprise"},
}


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


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


@router.post("/login", response_model=LoginResponse)
def login(req: LoginRequest, request: Request):
    user = fake_users_db.get(req.username)
    if not user or user["password"] != req.password:
        raise HTTPException(401, "Invalid credentials")

    session_id = create_session(
        user_id=user["user_id"],
        tier=user["tier"],
        metadata={
            "ip": request.client.host if request.client else "",
            "user_agent": request.headers.get("user-agent", ""),
            "device_name": request.headers.get("X-Device-Name", "Unknown"),
        }
    )

    token = create_jwt(user_id=user["user_id"], session_id=session_id)

    return LoginResponse(
        access_token=token,
        session_id=session_id,
        user_id=user["user_id"],
        tier=user["tier"],
    )


# To avoid circular imports, get_current_user is imported lazily
def _get_current_user():
    from app.main import get_current_user
    return get_current_user


@router.post("/logout")
def logout(user: dict = Depends(_get_current_user())):
    revoke_session(user["session_id"])
    return {"message": "Logged out from this device"}


@router.post("/logout-all")
def logout_all(user: dict = Depends(_get_current_user())):
    count = revoke_all_sessions(user["user_id"])
    return {"message": f"Logged out from {count} devices"}


@router.get("/sessions")
def my_sessions(user: dict = Depends(_get_current_user())):
    sessions = list_user_sessions(user["user_id"])
    for s in sessions:
        s["is_current"] = s["session_id"] == user["session_id"]
    return {"sessions": sessions, "total": len(sessions)}


@router.post("/sessions/{session_id}/revoke")
def revoke_specific(session_id: str, user: dict = Depends(_get_current_user())):
    if session_id == user["session_id"]:
        raise HTTPException(400, "Use /auth/logout to revoke current session")

    sessions = list_user_sessions(user["user_id"])
    if not any(s["session_id"] == session_id for s in sessions):
        raise HTTPException(404, "Session not found or not yours")

    revoke_session(session_id)
    return {"message": "Session revoked"}

app/routers/api.py

"""
API endpoints protected with a rate limit.
"""
from fastapi import APIRouter, Depends


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


def _get_current_user():
    from app.main import get_current_user
    return get_current_user


@router.get("/products")
def list_products(user: dict = Depends(_get_current_user())):
    """A general endpoint (rate limit: 100/hr free, 1000/hr pro, 10000/hr enterprise)."""
    return {
        "products": [{"id": i, "name": f"Product {i}"} for i in range(10)],
        "user_tier": user["tier"],
    }


@router.get("/search")
def search(q: str = "", user: dict = Depends(_get_current_user())):
    """An expensive endpoint (rate limit: 20/hr free, 200/hr pro, 2000/hr enterprise)."""
    return {
        "query": q,
        "results": [],
        "user_tier": user["tier"],
    }


@router.post("/orders")
def create_order(user: dict = Depends(_get_current_user())):
    """A write endpoint (rate limit: 10/hr free, 100/hr pro, 1000/hr enterprise)."""
    return {
        "order_id": "ord_abc123",
        "status": "created",
        "user_id": user["user_id"],
    }

app/routers/admin.py

"""
Admin endpoints (in production, protected with a role check).
"""
from fastapi import APIRouter, HTTPException
from app.auth import revoke_all_sessions, list_user_sessions


router = APIRouter(prefix="/admin", tags=["Admin"])


# In production: a dependency that verifies the admin role
# Simplified here (we assume it's an admin)


@router.post("/users/{user_id}/revoke")
def force_revoke_user(user_id: int):
    """Admin: force a global logout for a user (a compromised account, etc.)."""
    count = revoke_all_sessions(user_id)
    return {"message": f"Revoked {count} sessions for user {user_id}"}


@router.get("/users/{user_id}/sessions")
def admin_list_user_sessions(user_id: int):
    """Admin: view a user's active sessions."""
    sessions = list_user_sessions(user_id)
    return {"sessions": sessions, "total": len(sessions)}

app/routers/__init__.py

"""The rate limiter service's routers."""

app/__init__.py

"""Rate Limiter Service."""

Running it

# Setup
cd ~/projects/redis-guide/module-03-rate-limiting
mkdir -p rate-limiter-service && cd rate-limiter-service

# Create the structure
mkdir -p app/routers tests
touch app/__init__.py app/routers/__init__.py tests/__init__.py

# Copy the code from the files above
# (this assumes you already copied main.py, config.py, auth.py, rate_limiter.py, routers/*)

# Set up the venv
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

# Verify Redis
docker ps | grep redis

# Start the server
uvicorn app.main:app --reload --port 8000

Expected output:

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO:     Application startup complete.
2026-04-25 10:00:00 [main] INFO: Rate Limiter Service started

Test 1: The basic auth flow

# 1. Log in as Alice (the free tier)
TOKEN=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -H "X-Device-Name: MacBook" \
  -d '{"username": "alice", "password": "secret"}' | jq -r .access_token)

echo "Token obtained"

# 2. /me
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/me | jq
# {"user_id": 1, "tier": "free", "device": "MacBook", ...}

# 3. /api/products (it should go through)
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/products | jq

# 4. Check the headers
curl -s -i -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/products | grep -i "x-ratelimit"
# X-RateLimit-Limit: 100
# X-RateLimit-Remaining: 99
# X-RateLimit-Reset: 1714069320

Test 2: The rate limit per tier

# Free tier: 100/hr on /api/products
# Pro tier: 1000/hr
# Enterprise: 10000/hr

# Log in as a pro user
TOKEN_PRO=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "bob_pro", "password": "secret"}' | jq -r .access_token)

# Check the X-RateLimit-Limit header
curl -s -i -H "Authorization: Bearer $TOKEN_PRO" http://localhost:8000/api/products | grep -i "x-ratelimit-limit"
# X-RateLimit-Limit: 1000   ← higher than free

# Log in as enterprise
TOKEN_ENT=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "ent_carol", "password": "secret"}' | jq -r .access_token)

curl -s -i -H "Authorization: Bearer $TOKEN_ENT" http://localhost:8000/api/products | grep -i "x-ratelimit-limit"
# X-RateLimit-Limit: 10000

Test 3: A rate limit hit → 429

# A free user: 100/hr on /api/products
# Let's make 105 quick requests

for i in {1..105}; do
  STATUS=$(curl -s -o /dev/null -w "%{http_code}" \
    -H "Authorization: Bearer $TOKEN" \
    http://localhost:8000/api/products)
  if [ "$STATUS" != "200" ]; then
    echo "Request $i: $STATUS"
  fi
done

# Output:
# Request 101: 429
# Request 102: 429
# Request 103: 429
# Request 104: 429
# Request 105: 429

# See the 429's details
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/products | jq
# {
#   "error": "rate_limit_exceeded",
#   "message": "Rate limit exceeded for general...",
#   "retry_after_seconds": 3540,
#   ...
# }

Test 4: A concurrent load test (tests/test_rate_limit_load.py)

"""
A load test: 500 concurrent requests to verify atomicity.
"""
import asyncio
import httpx
import pytest


BASE_URL = "http://localhost:8000"


async def get_token():
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{BASE_URL}/auth/login",
            json={"username": "alice", "password": "secret"}
        )
        return r.json()["access_token"]


async def make_request(client, token, path):
    try:
        r = await client.get(
            f"{BASE_URL}{path}",
            headers={"Authorization": f"Bearer {token}"},
            timeout=10
        )
        return r.status_code
    except Exception as e:
        return None


@pytest.mark.asyncio
async def test_concurrent_rate_limiting():
    """
    Free tier: 100/hr on /api/products.
    500 concurrent requests should result in exactly 100 success and 400 rate-limited.
    """
    token = await get_token()

    async with httpx.AsyncClient() as client:
        tasks = [make_request(client, token, "/api/products") for _ in range(500)]
        results = await asyncio.gather(*tasks)

    counts = {}
    for status in results:
        counts[status] = counts.get(status, 0) + 1

    print(f"\nResults: {counts}")

    # Atomicity check: exactly 100 should pass
    assert counts.get(200, 0) == 100, f"Expected 100 successful, got {counts.get(200, 0)}"
    assert counts.get(429, 0) == 400, f"Expected 400 rate-limited, got {counts.get(429, 0)}"


if __name__ == "__main__":
    asyncio.run(test_concurrent_rate_limiting())

Run it:

# Make sure the server is running
# In another terminal:
pytest tests/test_rate_limit_load.py -v -s

Expected output:

Results: {200: 100, 429: 400}
PASSED

✅ Atomicity verified: with 500 concurrent requests, the rate limiter allowed EXACTLY 100 (the free tier's limit).

Test 5: The sessions flow

# Clean up Redis
docker exec redis-dev redis-cli FLUSHDB

# Log in from 3 devices
T1=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -H "X-Device-Name: MacBook Pro" \
  -d '{"username": "alice", "password": "secret"}' | jq -r .access_token)

T2=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -H "X-Device-Name: iPhone" \
  -d '{"username": "alice", "password": "secret"}' | jq -r .access_token)

T3=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -H "X-Device-Name: iPad" \
  -d '{"username": "alice", "password": "secret"}' | jq -r .access_token)

# List my sessions from T1
curl -s -H "Authorization: Bearer $T1" http://localhost:8000/auth/sessions | jq
# {"sessions": [3 entries], "total": 3}

# A logout-all from T1
curl -s -X POST -H "Authorization: Bearer $T1" http://localhost:8000/auth/logout-all
# {"message": "Logged out from 3 devices"}

# Verify: T2 no longer works
curl -s -H "Authorization: Bearer $T2" http://localhost:8000/me
# {"detail": "Session revoked or expired. Please login again."}

End-to-end verification (the complete script)

#!/bin/bash
# verify.sh - A complete smoke test

set -e
BASE="http://localhost:8000"

echo "=== 1. Health check ==="
curl -s "$BASE/health" | jq
echo

echo "=== 2. Login ==="
T=$(curl -s -X POST "$BASE/auth/login" \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "password": "secret"}' | jq -r .access_token)
[ -n "$T" ] && echo "✓ Token obtained" || (echo "✗ Login failed" && exit 1)

echo
echo "=== 3. /me ==="
curl -s -H "Authorization: Bearer $T" "$BASE/me" | jq

echo
echo "=== 4. /api/products (with the rate limit headers) ==="
curl -s -i -H "Authorization: Bearer $T" "$BASE/api/products" | head -10 | grep -i "x-ratelimit\|http"

echo
echo "=== 5. Sessions list ==="
curl -s -H "Authorization: Bearer $T" "$BASE/auth/sessions" | jq

echo
echo "=== 6. Logout ==="
curl -s -X POST -H "Authorization: Bearer $T" "$BASE/auth/logout" | jq

echo
echo "=== 7. /me after the logout (it should fail) ==="
curl -s -H "Authorization: Bearer $T" "$BASE/me" | jq

echo
echo "✅ Verification complete"

Run it:

chmod +x verify.sh
./verify.sh

Troubleshooting

Problem 1: The concurrent tests show a race condition (more than 100 200s)

Cause: The count > limit check in rate_limiter.py isn't 100% atomic.

Solution: Implement it with a Lua script (shown in capsule 03's troubleshooting). For local testing with 500 requests, atomicity typically holds; under real load with thousands of concurrent requests, consider Lua.

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

Cause: The middleware returns before the rate limit check if the JWT is invalid.

Solution: The rate limiting should always run BEFORE the auth check (even unauthenticated requests can be rate-limited by IP). This is already implemented correctly in the middleware.

Problem 3: The session renews but the JWT expires

Cause: The JWT TTL is 24h (absolute). After 24h, even if the session is still "active" in Redis, the JWT is invalid.

Solution: Implement refresh tokens (out of scope for this module, see guide #9). For now, after 24h the user has to log in again.

Problem 4: Redis goes down and the app stops responding

Cause: If the rate limit check has no try/except, the RedisError propagates.

Solution: Already implemented: the rate limiter returns degraded=True if Redis fails, and it allows the request. The X-RateLimit-Degraded: true header tells you.

Problem 5: MAX_SESSIONS_PER_USER isn't honored

Cause: A race condition in enforce_session_limit with sessions created concurrently.

Solution: Acceptable for production. The user could have 6 sessions for a moment, but the next operation caps them. If you need a strict guarantee, use a Lua script.

Problem 6: The tests fail in CI because of timing

Cause: In CI, the Redis container can take a while to be ready.

Solution: A healthcheck in CI:

# .github/workflows/test.yml
services:
  redis:
    image: redis:7
    options: >-
      --health-cmd "redis-cli ping"
      --health-interval 10s
      --health-timeout 5s
      --health-retries 5

Module 3 summary

You close module 3 with command of:

Rate limiting algorithms:

  • Token bucket: bursts allowed up to capacity
  • Leaky bucket: traffic smoothing
  • Sliding window with sorted sets (the professional one): ZADD/ZRANGEBYSCORE/ZREMRANGEBYSCORE

The fixed window bug:

  • An INCR + EXPIRE counter lets you double the limit by timing the reset
  • A sliding window mathematically avoids it

The granularity of rate limiting:

  • By IP (anti-DDoS)
  • By authenticated user (fair use, multi-tier)
  • By specific endpoint (protecting the expensive ones)
  • Combinable in defense-in-depth layers

Standard HTTP headers:

  • X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset
  • Retry-After on 429 responses
  • The 429 Too Many Requests status code

JWT + Redis sessions:

  • The JWT verifies identity quickly (stateless)
  • The Redis session allows immediate revocation
  • The professional pattern (GitHub, Slack, everyone)
  • A sliding TTL for active sessions
  • Local logout vs global logout

The Rate Limiter Service mini-project:

  • A complete multi-tier API with FastAPI
  • Sliding window middleware + JWT auth
  • A concurrent load test verifying atomicity
  • Graceful degradation when Redis fails
  • Structured logging for auditing

What's coming in module 4: Pub/Sub & FastAPI Integration. You'll move from synchronous operations to async, integrate Redis with FastAPI using dependency injection and connection pooling, and implement Pub/Sub for cache invalidation events between components.

Additional resources

  1. FastAPI Middleware — The official docs for the pattern we used
  2. HTTPx Async — An async HTTP client for load tests
  3. pytest-asyncio — A framework for async tests
  4. Redis Memory Optimization — Best practices for apps with many sessions/rate limits
  5. Stripe Engineering: Rate Limiters — The production case that inspired this module
  6. The Twelve-Factor App: Concurrency — Why rate limiting shared between workers matters

What's next?

You've closed Module 3: Rate Limiting & Sessions. In Module 4: Pub/Sub & FastAPI Integration you get into the last piece before the capstone project: async Redis with redis.asyncio (not the deprecated aioredis), Pub/Sub for messaging between components, professional dependency injection with FastAPI, connection pooling, and automatic caching middleware.

Before moving on, make sure you:

  • Have the Rate Limiter Service running
  • Have verified the auth flow (login, /me, logout, logout-all)
  • Have the load tests passing (100/500 with the free tier, correct atomicity)
  • See the X-RateLimit-* headers on every response
  • Understand why each architectural decision is where it is

If all 5 are ✅, you've completed module 3. On to module 4.

← PreviousSessions with Redis (complementing JWT)Next →Introduction to Module 4: Pub/Sub and FastAPI Integration