Module 5: Capstone Project — API with Caching Strategy

Capsule 04 — Sessions + Pub/Sub + Monitoring

Next →

Overview

In the previous capsule you left the project with working caching + rate limiting. Now you close out the business logic by adding the three blocks that are missing for the API to be production-ready:

  1. Sessions with JWT + Redis: real login, immediate logout (not waiting for the JWT to expire), logout-all (invalidating all the user's sessions), list devices.
  2. Write-through for PUT /api/users/{id}: when you update a user, you write to Postgres and to Redis in the same operation. The cache is always fresh, the write latency is a bit higher — the tradeoff you saw in Module 2.
  3. Write-behind for POST /analytics/event: the client gets an immediate 202 Accepted; an async worker drains the Redis queue into Postgres in batches. Maximum throughput, eventual durability.
  4. A Pub/Sub listener: a background task that listens on the cache:invalidate channel and processes events (logging, metrics, future cases like notifying external workers). This closes the distributed invalidation loop you saw in M4's capsule 02.
  5. A WebSocket bridge /ws/notifications: connected clients receive Redis Pub/Sub events in real time (you already built it in M4's capsule 05; here you integrate it into the project).
  6. Extended metrics in /metrics: on top of cache hits/misses, you now report active sessions, Pub/Sub events received, the write-behind queue length, and captured errors.

By the end of this capsule, the project has all the guide's features integrated. Capsule 05 is just the close: Docker Compose, end-to-end verification, a README, and a portfolio.

The tone: you keep building on what you already have. Each block references the module where you learned it and integrates it into the project. Without re-explaining the fundamentals.

JWT + Redis sessions

In capsule 03 you left a _get_user_and_tier() stub in the rate limit middleware that always returned None. It's time to replace it with a real JWT plus sessions in Redis.

The pattern (a quick reminder)

You saw it in full in Module 3's capsule 04. A one-line reminder: the JWT identifies the user, but the session lives in Redis (a HASH with metadata) and in a SET per user (user:{id}:sessions). Logout = DEL session:{sid} + SREM user:{id}:sessions {sid}. Logout-all = a pipeline of DELs over every sid in the SET.

app/auth/jwt_handler.py

Replace the stub file (if it existed) with the complete implementation.

# app/auth/jwt_handler.py
from datetime import datetime, timezone, timedelta
from typing import Optional

import jwt as pyjwt
from fastapi import HTTPException, status

from app.config import settings


def create_access_token(user_id: str, session_id: str, tier: str = "free") -> str:
    """Creates a JWT with the user_id, session_id, and tier."""
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "sid": session_id,
        "tier": tier,
        "iat": now,
        "exp": now + timedelta(seconds=settings.JWT_EXPIRES_SECONDS),
    }
    return pyjwt.encode(payload, settings.JWT_SECRET, algorithm="HS256")


def decode_access_token(token: str) -> dict:
    """Decodes a JWT. It raises an HTTPException if it's invalid or expired."""
    try:
        payload = pyjwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
        return payload
    except pyjwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except pyjwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )

You need two settings in app/config.py (add them if you don't have them):

# app/config.py — add
JWT_SECRET: str = "change-me-in-production"  # In a real .env
JWT_EXPIRES_SECONDS: int = 3600  # 1h
SESSION_TTL_SECONDS: int = 86400  # 24h (longer than the JWT)

app/auth/sessions.py

Sessions as a HASH + a SET per user. The pattern is identical to M3's capsule 04, adapted to the project.

# app/auth/sessions.py
import secrets
from datetime import datetime, timezone
from typing import Optional

from redis.asyncio import Redis

from app.config import settings


def _session_key(session_id: str) -> str:
    return f"session:{session_id}"


def _user_sessions_key(user_id: str) -> str:
    return f"user:{user_id}:sessions"


async def create_session(
    r: Redis,
    user_id: str,
    tier: str,
    user_agent: str = "",
    ip: str = "",
) -> str:
    """Creates a session in Redis. It returns the session_id."""
    session_id = secrets.token_urlsafe(32)
    now = datetime.now(timezone.utc).isoformat()

    pipe = r.pipeline()
    pipe.hset(_session_key(session_id), mapping={
        "user_id": user_id,
        "tier": tier,
        "created_at": now,
        "last_seen": now,
        "user_agent": user_agent,
        "ip": ip,
    })
    pipe.expire(_session_key(session_id), settings.SESSION_TTL_SECONDS)
    pipe.sadd(_user_sessions_key(user_id), session_id)
    pipe.expire(_user_sessions_key(user_id), settings.SESSION_TTL_SECONDS)
    await pipe.execute()

    return session_id


async def get_session(r: Redis, session_id: str) -> Optional[dict]:
    """Returns the session's metadata, or None if it doesn't exist/expired."""
    data = await r.hgetall(_session_key(session_id))
    return data if data else None


async def touch_session(r: Redis, session_id: str) -> None:
    """Updates last_seen and extends the TTL (a sliding session)."""
    now = datetime.now(timezone.utc).isoformat()
    pipe = r.pipeline()
    pipe.hset(_session_key(session_id), "last_seen", now)
    pipe.expire(_session_key(session_id), settings.SESSION_TTL_SECONDS)
    await pipe.execute()


async def revoke_session(r: Redis, session_id: str) -> bool:
    """Deletes a specific session. It returns True if it existed."""
    data = await r.hgetall(_session_key(session_id))
    if not data:
        return False

    user_id = data.get("user_id")
    pipe = r.pipeline()
    pipe.delete(_session_key(session_id))
    if user_id:
        pipe.srem(_user_sessions_key(user_id), session_id)
    await pipe.execute()
    return True


async def revoke_all_user_sessions(r: Redis, user_id: str) -> int:
    """Logout-all: deletes every session the user has."""
    sids = await r.smembers(_user_sessions_key(user_id))
    if not sids:
        return 0

    pipe = r.pipeline()
    for sid in sids:
        pipe.delete(_session_key(sid))
    pipe.delete(_user_sessions_key(user_id))
    await pipe.execute()
    return len(sids)


async def list_user_sessions(r: Redis, user_id: str) -> list[dict]:
    """Returns the metadata of every active session the user has."""
    sids = await r.smembers(_user_sessions_key(user_id))
    if not sids:
        return []

    sessions = []
    for sid in sids:
        data = await r.hgetall(_session_key(sid))
        if data:
            sessions.append({"session_id": sid, **data})
    return sessions

app/auth/dependencies.py

This is the dependency you'll use in endpoints that require auth. It verifies the JWT and that the session exists in Redis (this is what makes immediate logout possible).

# app/auth/dependencies.py
from typing import Annotated

from fastapi import Depends, HTTPException, status, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from redis.asyncio import Redis

from app.dependencies import get_redis
from app.auth.jwt_handler import decode_access_token
from app.auth.sessions import get_session, touch_session

security = HTTPBearer()


async def get_current_user(
    request: Request,
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
    r: Annotated[Redis, Depends(get_redis)],
) -> dict:
    """
    Verifies the JWT + the session in Redis. It returns a dict with user_id, tier, session_id.
    If the JWT is valid but the session was revoked, it returns a 401 (immediate logout).
    """
    payload = decode_access_token(credentials.credentials)
    user_id = payload.get("sub")
    session_id = payload.get("sid")
    tier = payload.get("tier", "free")

    if not user_id or not session_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token payload",
        )

    session = await get_session(r, session_id)
    if not session:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Session revoked or expired",
        )

    # A sliding session: it extends the TTL on every authenticated request
    await touch_session(r, session_id)

    # So the rate limit middleware can read it (request.state)
    request.state.user_id = user_id
    request.state.tier = tier
    request.state.session_id = session_id

    return {"user_id": user_id, "tier": tier, "session_id": session_id}

app/routers/auth.py

Login, logout, logout-all, and a sessions list. In this version the "login" doesn't validate a password against a users DB — it uses a hardcoded user for simplicity (the guide is about Redis, not full auth). If you wanted a real DB, you'd swap the check for the verify_password() from your Auth guide (#9).

# app/routers/auth.py
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status, Request
from pydantic import BaseModel
from redis.asyncio import Redis

from app.dependencies import get_redis
from app.auth.jwt_handler import create_access_token
from app.auth.sessions import (
    create_session,
    revoke_session,
    revoke_all_user_sessions,
    list_user_sessions,
)
from app.auth.dependencies import get_current_user

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


class LoginRequest(BaseModel):
    user_id: str  # In real life: email + password
    tier: str = "free"  # free, pro, enterprise


class LoginResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
    session_id: str


# Demo users (in production this would come from Postgres)
DEMO_USERS = {
    "alice": "free",
    "bob": "pro",
    "carol": "enterprise",
}


@router.post("/login", response_model=LoginResponse)
async def login(
    payload: LoginRequest,
    request: Request,
    r: Annotated[Redis, Depends(get_redis)],
):
    """A demo login. In production: verify the password against a users DB."""
    if payload.user_id not in DEMO_USERS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found",
        )

    tier = DEMO_USERS[payload.user_id]
    user_agent = request.headers.get("user-agent", "")
    ip = request.client.host if request.client else ""

    session_id = await create_session(r, payload.user_id, tier, user_agent, ip)
    token = create_access_token(payload.user_id, session_id, tier)

    return LoginResponse(access_token=token, session_id=session_id)


@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
async def logout(
    user: Annotated[dict, Depends(get_current_user)],
    r: Annotated[Redis, Depends(get_redis)],
):
    """Logs out of the current session."""
    await revoke_session(r, user["session_id"])


@router.post("/logout-all", status_code=status.HTTP_204_NO_CONTENT)
async def logout_all(
    user: Annotated[dict, Depends(get_current_user)],
    r: Annotated[Redis, Depends(get_redis)],
):
    """Logs out of every session the user has (on every device)."""
    await revoke_all_user_sessions(r, user["user_id"])


@router.get("/sessions")
async def list_my_sessions(
    user: Annotated[dict, Depends(get_current_user)],
    r: Annotated[Redis, Depends(get_redis)],
):
    """Lists every active session the current user has."""
    sessions = await list_user_sessions(r, user["user_id"])
    return {"count": len(sessions), "sessions": sessions}

Updating the rate limit middleware

Replace the _get_user_and_tier() stub in app/rate_limit/middleware.py with something that reads from the JWT. Since the middleware runs before the dependencies, decode the token yourself (without verifying the session yet — get_current_user does that afterwards).

# app/rate_limit/middleware.py — only the _get_user_and_tier method
import jwt as pyjwt
from app.config import settings


async def _get_user_and_tier(self, request: Request) -> tuple[Optional[str], str]:
    """Extracts the user_id and tier from the JWT. With no token, it returns (None, 'free')."""
    auth = request.headers.get("authorization", "")
    if not auth.startswith("Bearer "):
        return None, "free"

    token = auth[7:]
    try:
        payload = pyjwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
        return payload.get("sub"), payload.get("tier", "free")
    except pyjwt.InvalidTokenError:
        return None, "free"

Why the middleware doesn't verify the session: if it did, every request would hit Redis twice (the session + the rate limit). Better: the middleware only reads the tier for rate limiting; the dependencies verify the session in the endpoints that need it. An accepted trade-off.

Write-through: PUT /api/users/{id}

A reminder of the pattern (M2's capsule 03): write-through writes to the DB and the cache in the same operation. The guarantee: the cache is always consistent. The cost: slightly higher write latency, and if the cache fails, the write can degrade.

In the project, /api/users/{id} already has cache-aside on the GET (you built it in capsule 03). Now you add write-through on the PUT.

# app/routers/users.py — add the PUT
import json
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_redis, get_db
from app.auth.dependencies import get_current_user
from app.models import User, UserUpdate
from app.config import settings


@router.put("/{user_id}", response_model=User)
async def update_user(
    user_id: str,
    payload: UserUpdate,
    db: Annotated[AsyncSession, Depends(get_db)],
    r: Annotated[Redis, Depends(get_redis)],
    current: Annotated[dict, Depends(get_current_user)],
):
    """Write-through: it updates Postgres + the cache + publishes an invalidation."""
    if current["user_id"] != user_id and current["tier"] != "enterprise":
        raise HTTPException(status_code=403, detail="Forbidden")

    # 1. Update the DB (the source of truth)
    user = await db.get(UserORM, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    for field, value in payload.model_dump(exclude_unset=True).items():
        setattr(user, field, value)

    await db.commit()
    await db.refresh(user)

    user_dict = User.model_validate(user).model_dump(mode="json")

    # 2. Write-through: update the cache atomically
    cache_key = f"user:{user_id}"
    pipe = r.pipeline()
    pipe.delete(cache_key)
    pipe.hset(cache_key, mapping={k: json.dumps(v) for k, v in user_dict.items()})
    pipe.expire(cache_key, settings.CACHE_TTL_USER)
    await pipe.execute()

    # 3. Publish an event to invalidate the related caches (the users list, etc.)
    await r.publish("cache:invalidate", json.dumps({
        "type": "user.updated",
        "user_id": user_id,
        "keys": [f"users:list:*"],  # The pattern to invalidate in the related caches
    }))

    return user

Why the DELETE before the HSET: if the user had an optional field with a value before (e.g., phone) and it's now None, the HSET doesn't delete it automatically. The preceding DELETE cleans it up. A small optimization: if you guarantee that UserUpdate always includes every field, you can drop the DELETE.

Why the publish comes after the commit: if the commit fails, you don't want to have already notified the other workers. The order: DB → cache → publish.

Write-behind: POST /analytics/event

The opposite pattern. The client gets an immediate 202 Accepted, the event is enqueued in Redis (a LIST), and a worker drains it in batches to Postgres. Maximum throughput. The risk: eventual durability (if Redis goes down before the flush, the queued events are lost).

You saw it theoretically in M2's capsule 03. Here you apply it.

The endpoint

# app/routers/analytics.py
import json
from datetime import datetime, timezone
from typing import Annotated

from fastapi import APIRouter, Depends, status, Request
from pydantic import BaseModel
from redis.asyncio import Redis

from app.dependencies import get_redis


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


class AnalyticsEvent(BaseModel):
    event_type: str  # "page_view", "click", "purchase", etc.
    user_id: str | None = None
    properties: dict = {}


QUEUE_KEY = "analytics:queue"


@router.post("/event", status_code=status.HTTP_202_ACCEPTED)
async def track_event(
    event: AnalyticsEvent,
    request: Request,
    r: Annotated[Redis, Depends(get_redis)],
):
    """Write-behind: it enqueues in Redis, the worker drains it later."""
    payload = {
        "event_type": event.event_type,
        "user_id": event.user_id,
        "properties": event.properties,
        "ip": request.client.host if request.client else None,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    await r.rpush(QUEUE_KEY, json.dumps(payload))
    return {"status": "accepted", "queue_length": await r.llen(QUEUE_KEY)}

The worker

A background task that starts in main.py's lifespan. It drains the queue in batches with an atomic LRANGE + LTRIM via a pipeline.

# app/workers/analytics_worker.py
import asyncio
import json
import logging
from typing import TYPE_CHECKING

from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import async_session_maker
from app.redis_client import get_pool
from app.config import settings

if TYPE_CHECKING:
    from redis.asyncio import Redis

logger = logging.getLogger("analytics_worker")

QUEUE_KEY = "analytics:queue"
BATCH_SIZE = 100
FLUSH_INTERVAL = 5  # seconds


async def _flush_batch(r: "Redis") -> int:
    """Reads up to BATCH_SIZE events from the queue and writes them to Postgres."""
    pipe = r.pipeline()
    pipe.lrange(QUEUE_KEY, 0, BATCH_SIZE - 1)
    pipe.ltrim(QUEUE_KEY, BATCH_SIZE, -1)
    raw_events, _ = await pipe.execute()

    if not raw_events:
        return 0

    events = [json.loads(e) for e in raw_events]

    async with async_session_maker() as db:
        try:
            from app.db import AnalyticsEventORM
            await db.execute(
                insert(AnalyticsEventORM),
                [{
                    "event_type": e["event_type"],
                    "user_id": e.get("user_id"),
                    "properties": e.get("properties", {}),
                    "ip": e.get("ip"),
                    "timestamp": e["timestamp"],
                } for e in events]
            )
            await db.commit()
            logger.info(f"Flushed {len(events)} analytics events to DB")
            return len(events)
        except Exception as ex:
            await db.rollback()
            # Re-enqueue at the end so we don't lose the data (best-effort)
            for e in events:
                await r.rpush(QUEUE_KEY, json.dumps(e))
            logger.error(f"Flush failed, re-queued {len(events)} events: {ex}")
            return 0


async def analytics_worker_loop():
    """An infinite loop: every FLUSH_INTERVAL seconds, it drains a batch."""
    from redis.asyncio import Redis
    pool = get_pool()
    r = Redis(connection_pool=pool)

    logger.info("Analytics worker started")
    try:
        while True:
            try:
                await _flush_batch(r)
            except Exception as ex:
                logger.exception(f"Worker iteration failed: {ex}")
            await asyncio.sleep(FLUSH_INTERVAL)
    except asyncio.CancelledError:
        logger.info("Analytics worker stopping, flushing remaining...")
        # A final flush before exiting
        try:
            await _flush_batch(r)
        except Exception:
            logger.exception("Final flush failed")
        raise

Integrating it in main.py

# app/main.py — update the lifespan
import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.workers.analytics_worker import analytics_worker_loop


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    init_pool(settings.REDIS_URL, max_connections=50)

    # Background workers
    worker_task = asyncio.create_task(analytics_worker_loop())
    pubsub_task = asyncio.create_task(pubsub_listener_loop())  # see the next section

    yield

    # Shutdown
    worker_task.cancel()
    pubsub_task.cancel()
    try:
        await asyncio.gather(worker_task, pubsub_task, return_exceptions=True)
    except Exception:
        pass

    await close_pool()

And if the server crashes with events in the queue? They stay in Redis. When the server restarts, the worker drains them. The real risk: if Redis goes down with the events in memory. That's why write-behind is for data where "losing a handful" is acceptable (analytics, logs). NEVER for payments or transactions.

The Pub/Sub listener

In M4 you saw how Redis Pub/Sub enables distributed invalidation. Here you integrate it into the project: a background task that listens on cache:invalidate and acts (logging + metrics + optionally deleting local caches).

# app/pubsub/listener.py
import asyncio
import json
import logging
from typing import TYPE_CHECKING

from app.redis_client import get_pool
from app.metrics import metrics

if TYPE_CHECKING:
    from redis.asyncio import Redis

logger = logging.getLogger("pubsub_listener")


async def _handle_invalidation(event: dict, r: "Redis") -> None:
    """Processes a cache:invalidate event."""
    event_type = event.get("type")
    metrics["pubsub_events_received"] += 1
    metrics["pubsub_by_type"][event_type] = metrics["pubsub_by_type"].get(event_type, 0) + 1

    logger.info(f"Pub/Sub event: {event_type} → {event}")

    # Act based on the type
    keys_to_invalidate = event.get("keys", [])
    for key_pattern in keys_to_invalidate:
        if "*" in key_pattern:
            # A pattern: use SCAN so we don't block
            cursor = 0
            count = 0
            while True:
                cursor, batch = await r.scan(cursor, match=key_pattern, count=100)
                if batch:
                    await r.delete(*batch)
                    count += len(batch)
                if cursor == 0:
                    break
            if count:
                logger.info(f"Invalidated {count} keys matching {key_pattern}")
        else:
            await r.delete(key_pattern)


async def pubsub_listener_loop():
    """Subscribes to cache:invalidate and processes the events. An infinite loop."""
    from redis.asyncio import Redis
    pool = get_pool()
    r = Redis(connection_pool=pool)
    pubsub = r.pubsub()

    await pubsub.subscribe("cache:invalidate")
    logger.info("Pub/Sub listener subscribed to cache:invalidate")

    try:
        async for message in pubsub.listen():
            if message["type"] != "message":
                continue
            try:
                event = json.loads(message["data"])
                await _handle_invalidation(event, r)
            except json.JSONDecodeError:
                logger.warning(f"Invalid JSON in pub/sub: {message['data']!r}")
            except Exception:
                logger.exception("Failed to handle invalidation event")
    except asyncio.CancelledError:
        logger.info("Pub/Sub listener stopping")
        await pubsub.unsubscribe("cache:invalidate")
        await pubsub.close()
        raise

Why a separate listener if the routers already delete the caches directly? Because it enables horizontal scaling: if you have 3 instances of the API running, one does a PUT and publishes → the other 2 listen and delete their local caches (if they had any). In the current project the cache is Redis (shared), so the listener is more for observability + future extension.

The WebSocket bridge /ws/notifications

You already built the complete pattern in M4's capsule 05 (Real-time Notifications). Here you integrate it into the project. You reuse the same ConnectionManager and bridge.

# app/routers/websocket.py
import asyncio
import json
import logging
from typing import Annotated

from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Query, HTTPException
from redis.asyncio import Redis

from app.dependencies import get_redis
from app.auth.jwt_handler import decode_access_token
from app.auth.sessions import get_session

logger = logging.getLogger("ws")

router = APIRouter()


class ConnectionManager:
    def __init__(self):
        self.active: dict[str, set[WebSocket]] = {}  # user_id → set of WebSockets

    async def connect(self, user_id: str, ws: WebSocket):
        await ws.accept()
        self.active.setdefault(user_id, set()).add(ws)

    def disconnect(self, user_id: str, ws: WebSocket):
        if user_id in self.active:
            self.active[user_id].discard(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: str, message: dict):
        if user_id not in self.active:
            return
        dead = []
        for ws in self.active[user_id]:
            try:
                await ws.send_json(message)
            except Exception:
                dead.append(ws)
        for ws in dead:
            self.active[user_id].discard(ws)


manager = ConnectionManager()


@router.websocket("/ws/notifications")
async def notifications_ws(
    websocket: WebSocket,
    token: str = Query(...),
):
    """A WebSocket authenticated by a JWT in the query string. It receives the user's events."""
    try:
        payload = decode_access_token(token)
        user_id = payload["sub"]
        session_id = payload["sid"]
    except HTTPException:
        await websocket.close(code=1008, reason="Invalid token")
        return

    # Verify the session is still valid
    from app.redis_client import get_pool
    from redis.asyncio import Redis
    pool = get_pool()
    r = Redis(connection_pool=pool)
    session = await get_session(r, session_id)
    if not session:
        await websocket.close(code=1008, reason="Session revoked")
        return

    await manager.connect(user_id, websocket)
    logger.info(f"WS connected: user={user_id}")

    try:
        while True:
            # Keep the connection alive. If you want to allow client → server messages, parse them here.
            await websocket.receive_text()
    except WebSocketDisconnect:
        logger.info(f"WS disconnected: user={user_id}")
    finally:
        manager.disconnect(user_id, websocket)

The bridge: Pub/Sub → WebSocket

Extend pubsub_listener_loop so it also pushes to connected WebSocket clients:

# app/pubsub/listener.py — add at the end of _handle_invalidation
from app.routers.websocket import manager as ws_manager


async def _notify_websocket_users(event: dict) -> None:
    """If the event has a user_id, notify the user via WS if they're connected."""
    user_id = event.get("user_id")
    if not user_id:
        return
    await ws_manager.send_to_user(user_id, {
        "type": event.get("type", "notification"),
        "data": event,
    })


# In _handle_invalidation, before the return:
async def _handle_invalidation(event: dict, r: "Redis") -> None:
    # ... the existing code ...
    await _notify_websocket_users(event)

Now when someone does a PUT /api/users/{id}, the Pub/Sub event reaches the listener, which invalidates the caches and notifies the user via WS if they're connected. The loop is complete.

Extended metrics: /metrics

app/metrics.py (if it doesn't exist) and the /metrics endpoint:

# app/metrics.py
from collections import defaultdict

metrics = {
    "cache_hits": 0,
    "cache_misses": 0,
    "cache_bypass": 0,
    "rate_limited": 0,
    "pubsub_events_received": 0,
    "pubsub_by_type": defaultdict(int),
    "errors_by_type": defaultdict(int),
}


def reset_metrics():
    metrics["cache_hits"] = 0
    metrics["cache_misses"] = 0
    metrics["cache_bypass"] = 0
    metrics["rate_limited"] = 0
    metrics["pubsub_events_received"] = 0
    metrics["pubsub_by_type"].clear()
    metrics["errors_by_type"].clear()
# app/main.py — the /metrics endpoint
from app.metrics import metrics


@app.get("/metrics", tags=["monitoring"])
async def get_metrics(r: Redis = Depends(get_redis)):
    """The system's metrics. In production you'd use Prometheus."""
    info = await r.info("clients")
    queue_length = await r.llen("analytics:queue")

    # Active sessions: an approximation by counting keys
    session_count = 0
    cursor = 0
    while True:
        cursor, batch = await r.scan(cursor, match="session:*", count=500)
        session_count += len(batch)
        if cursor == 0:
            break

    return {
        "cache": {
            "hits": metrics["cache_hits"],
            "misses": metrics["cache_misses"],
            "bypass": metrics["cache_bypass"],
            "hit_ratio": (
                metrics["cache_hits"]
                / max(metrics["cache_hits"] + metrics["cache_misses"], 1)
            ),
        },
        "rate_limit": {
            "blocked": metrics["rate_limited"],
        },
        "pubsub": {
            "events_received": metrics["pubsub_events_received"],
            "by_type": dict(metrics["pubsub_by_type"]),
        },
        "sessions": {
            "active": session_count,
        },
        "analytics": {
            "queue_length": queue_length,
        },
        "redis": {
            "connected_clients": info.get("connected_clients", 0),
        },
        "errors": dict(metrics["errors_by_type"]),
    }

Incrementing the counters

In each middleware/handler, increment the corresponding counter. Examples:

# CacheMiddleware
metrics["cache_hits"] += 1  # or _misses, or _bypass

# RateLimitMiddleware (when it blocks)
metrics["rate_limited"] += 1

# The listener (already integrated above)

# Any exception handler
metrics["errors_by_type"][type(exc).__name__] += 1

Partial verification

Bring up the complete API and test the end-to-end flow of auth + caching + invalidation + WebSocket.

# 1. Login
curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"user_id": "alice", "tier": "free"}' | jq

# Save the token in a variable
TOKEN=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"user_id": "alice", "tier": "free"}' | jq -r .access_token)

# 2. List the sessions
curl -s http://localhost:8000/auth/sessions \
  -H "Authorization: Bearer $TOKEN" | jq

# 3. PUT a user (write-through)
curl -s -X PUT http://localhost:8000/api/users/alice \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice Updated"}' | jq

# Verify the cache updated (the next GET should be a HIT with the new data)
curl -i http://localhost:8000/api/users/alice -H "Authorization: Bearer $TOKEN"

# 4. POST an analytics event (write-behind)
for i in {1..50}; do
  curl -s -X POST http://localhost:8000/analytics/event \
    -H "Content-Type: application/json" \
    -d "{\"event_type\": \"click\", \"user_id\": \"alice\", \"properties\": {\"page\": \"/home\"}}" > /dev/null
done

# Check the queue length (it should go down as the worker drains it)
redis-cli LLEN analytics:queue
sleep 6
redis-cli LLEN analytics:queue  # It should be 0 after the flush

# 5. Logout (the session is revoked immediately)
curl -X POST http://localhost:8000/auth/logout \
  -H "Authorization: Bearer $TOKEN"

# The next request with the same token: 401
curl -i http://localhost:8000/auth/sessions \
  -H "Authorization: Bearer $TOKEN"
# HTTP/1.1 401 Unauthorized
# {"detail": "Session revoked or expired"}

# 6. Metrics
curl -s http://localhost:8000/metrics | jq

# 7. WebSocket (in another terminal)
TOKEN2=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"user_id": "bob", "tier": "pro"}' | jq -r .access_token)

websocat "ws://localhost:8000/ws/notifications?token=$TOKEN2"
# Leave it open. From another terminal: PUT /api/users/bob → bob receives a notification

The expected result:

  • Login returns a token and a session_id
  • /auth/sessions lists the active sessions with their metadata (user-agent, ip, last_seen)
  • The PUT to a user updates Postgres + the cache + publishes an event → the next GET is a HIT with fresh data
  • The POST to analytics queues fast (200 OK <10 ms), and the worker drains it every 5s
  • Logout invalidates the session immediately → the next request is a 401 (it doesn't wait for the JWT to expire)
  • The metrics report active sessions, the queue length, hits/misses, and pub/sub events
  • The WebSocket receives a user.updated event when a PUT is made to the connected user

Troubleshooting

1. "Session revoked or expired" immediately after login The session was created (check it with redis-cli HGETALL session:<sid>), but the dependency is looking for a different sid. The cause: the JWT and the session use different ids. Check that create_access_token(user_id, session_id, ...) receives the real session_id returned by create_session().

2. The analytics worker doesn't drain the queue Check the logs: Analytics worker started should show up at boot. If it doesn't, the lifespan isn't creating the task. Check that app.add_lifespan(lifespan) comes before the routers in main.py.

3. The Pub/Sub listener receives events but doesn't process them async for message in pubsub.listen() also receives subscribe/unsubscribe events. That's why there's the if message["type"] != "message": continue. If you remove that check, it tries to parse JSON on messages that aren't data and crashes.

4. The WebSocket closes immediately Common causes: an invalid token (1008), or the client not waiting for the handshake. Try websocat -v to see the exact codes. Also: if your reverse proxy (nginx) doesn't have proxy_set_header Upgrade $http_upgrade, the WS upgrade fails.

5. A race condition: a PUT to a user, and an immediate GET returns old data If you see it consistently, there's an ordering bug. Verify that the cache's pipe.execute() completes before the response is returned. If you use parallel tasks, there's a risk of reading the old cache. The implementation above is sequential → correct.

6. The analytics queue grows and never drains The worker is crashing. Check the logs. A common cause: a schema mismatch between the event's JSON and the ORM column. Fix it by logging the first event that fails and comparing it against the AnalyticsEventORM model.

7. The metrics don't update metrics is a global dict in app/metrics.py. If you import it correctly in each module, they share state. If you redefine it in some module (from x import metrics; metrics = {...}), you create a new dict and break the sharing.

Recap

Sessions:

  • The JWT holds the user_id, session_id, and tier
  • The session lives in a Redis HASH + a SET per user
  • The get_current_user dependency verifies the JWT + the Redis session (immediate logout)
  • A sliding session: the TTL extends on every request

Write-through (PUT users):

  • The DB first (the source of truth)
  • The cache atomically via a pipeline (DELETE + HSET + EXPIRE)
  • A publish to notify the other workers/instances

Write-behind (POST analytics):

  • The client: an immediate 202 Accepted
  • A queue in a Redis LIST
  • An async worker drains it every 5s in batches of 100
  • A best-effort re-queue if the DB fails
  • A final flush on shutdown

The Pub/Sub listener:

  • A background task subscribed to cache:invalidate
  • It processes events: invalidates keys, records metrics, notifies WS
  • Cleanly cancelable with asyncio.CancelledError

The WebSocket bridge:

  • Auth with a JWT in the query string + a Redis session check
  • A ConnectionManager per user_id
  • The Pub/Sub listener pushes events to connected users

Monitoring:

  • /metrics: cache hits/misses, rate limited, active sessions, queue length, pub/sub events
  • Global counters in app/metrics.py, shared via import

The critical part: the project now integrates ALL the patterns from Modules 1-4. Capsule 05 is just the professional close: Docker Compose, end-to-end verification, a README, a portfolio.

Additional resources

  1. PyJWT Docs — The official JWT library
  2. FastAPI Background Tasks vs Workers — When to use each one
  3. Redis Pub/Sub limitations — Fire-and-forget, no persistence
  4. WebSocket Authentication Patterns — Query string vs subprotocols vs cookies
  5. Async Workers in FastAPI — Lifespan events
  6. Prometheus Python Client — For real production (this /metrics is educational)
  7. Cache-Aside vs Write-Through vs Write-Behind — The AWS comparison docs

What's next?

In Capsule 05 you close out the entire guide: the Docker Compose stack (FastAPI + Postgres + Redis), a verify.sh script that tests everything end-to-end, tests with pytest-asyncio, a professional README for your portfolio, and a closing summary with a bridge to your next guide (PostgreSQL & SQLAlchemy #8).

The project already works locally. Capsule 05 leaves it shipping-ready: anyone clones the repo, runs docker-compose up, and everything comes up.

Let's go.

← PreviousImplementation: Caching + Rate LimitingNext →Capsule 05 — End-to-end verification + Docker Compose + README + Portfolio