Sessions with Redis (complementing JWT)

Overview

You learned professional rate limiting with a sliding window. You close this module with the other critical topic: session management. Specifically, how Redis complements JWT to solve a fundamental limitation any API in production faces — the impossibility of revoking a JWT before it expires.

If you're coming from guide #9 (Authentication & Authorization), you know JWT: a token signed by the server, containing the user's claims, sent by the client on every request, validated by the server without touching the DB. It's elegant, scalable, and stateless. But it has a fundamental problem: once issued, a JWT can't be revoked until it expires. If a user loses their laptop with the session open, or if they report their account as compromised, the token stays valid until its natural TTL. If the TTL is 24 hours, that's 24 hours of exposure.

The professional solution is NOT "sessions instead of JWT" — it's JWT + Redis sessions together. The JWT verifies identity quickly and statelessly. The Redis session holds the mutable state: current roles, permissions, an is_revoked flag you can flip instantly. Every request verifies the JWT (no DB) + verifies that the session exists in Redis (1 ms). If you revoke the session in Redis, the next request fails automatically. It's immediate revocation without sacrificing performance.

This capsule covers the full implementation: how to store sessions with HSET, a sliding TTL to keep active sessions alive, local logout vs global logout ("log out on every device"), listing a user's active devices, and standard security patterns. By the end you have the complete toolkit for session management — the pattern GitHub, Slack, Notion, and every serious API uses.

The problem: JWT without revocation

A quick recap of the JWT flow:

1. POST /login {username, password}
   → The server validates the credentials against the DB
   → The server signs a JWT with its secret + claims (user_id, exp, etc.)
   → The server returns the JWT to the client

2. GET /products (with the header Authorization: Bearer <jwt>)
   → The server verifies the JWT's signature (no DB, ultra-fast)
   → The server reads the claims and knows that user_id=42
   → The server processes the request

3. The JWT expires after 24 hours → the client has to log in again

The good: stateless, scalable, with no DB queries for auth.

The bad: once the JWT is issued, there's no way to revoke it. It can only expire naturally.

Scenarios where you need immediate revocation

Scenario 1: A compromised account

The user reports: "My credentials were stolen, someone has access to my account."

Without sessions: the attacker has access until the token expires (hours to days). With sessions: DELETE session:abc123 = the attacker is logged out on their next request.

Scenario 2: An employee leaves the company

An employee leaves the company, and you strip their permissions in the system.

Without sessions: the JWT with the old permissions stays valid until it expires. With sessions: you revoke the session, and the next request with the old JWT fails because it doesn't find a session.

Scenario 3: A user changes their password

A user suspects their password was leaked and changes it.

Without sessions: the active sessions with the previous JWT keep working (until they expire). With sessions: you invalidate all the user's sessions, and every existing session dies.

Scenario 4: "Logout from all devices"

A common feature in modern apps (Gmail, Twitter, Notion).

Without sessions: impossible. With sessions: DELETE every session for the user_id.

The pattern: JWT + a Redis Session

How it works

1. POST /login
   → Validate the credentials
   → Create a JWT with the session_id included in its claims
   → Create the session in Redis (HSET session:{session_id})
   → Add the session_id to the user's set of sessions (SADD user:{user_id}:sessions)
   → Return the JWT to the client

2. GET /products (with the JWT)
   → Verify the JWT's signature (no DB)
   → Extract the session_id from the JWT
   → Verify the session exists in Redis (HEXISTS session:{session_id})
   → If it exists: read the up-to-date roles/permissions from Redis
   → If it does NOT exist: 401 Unauthorized (the session was revoked)

3. POST /logout
   → DELETE session:{session_id}
   → SREM user:{user_id}:sessions session_id
   → The client discards the JWT

4. POST /admin/revoke-user/{user_id}
   → SMEMBERS user:{user_id}:sessions → all the user's sessions
   → DELETE each one
   → DELETE user:{user_id}:sessions
   → The result: the user is logged out on EVERY device instantly

The division of responsibilities

The JWT stays stateless for fast verification. Redis is queried ONLY to check for revocation + read the mutable state. Every request is: 1 JWT verification (~0 ms) + 1 Redis lookup (~1 ms) = <2 ms total. Faster than going to PostgreSQL on every request.

Implementation

Setup

cd ~/projects/redis-guide/module-03-rate-limiting
mkdir sessions-demo && cd sessions-demo
python -m venv .venv
source .venv/bin/activate
pip install redis fastapi uvicorn pyjwt python-multipart

Session storage helpers

Create sessions.py:

"""
Session management with Redis.
"""
import time
import secrets
import json
import redis


r = redis.Redis(host='localhost', port=6379, decode_responses=True)


SESSION_TTL_SECONDS = 30 * 60  # 30 minutes
MAX_SESSIONS_PER_USER = 5      # the max number of concurrent devices


def create_session(user_id: int, metadata: dict | None = None) -> str:
    """
    Creates a new session in Redis.
    Returns: session_id
    """
    session_id = secrets.token_urlsafe(24)  # ~192 bits of entropy
    metadata = metadata or {}

    pipe = r.pipeline()

    # 1. Store the session's data as a hash
    session_data = {
        "user_id": str(user_id),
        "created_at": str(time.time()),
        "last_accessed": str(time.time()),
        "ip": metadata.get("ip", ""),
        "user_agent": metadata.get("user_agent", ""),
        "device_name": metadata.get("device_name", "Unknown device"),
    }
    pipe.hset(f"session:{session_id}", mapping=session_data)
    pipe.expire(f"session:{session_id}", SESSION_TTL_SECONDS)

    # 2. Add the session_id to the user's set of sessions
    pipe.sadd(f"user:{user_id}:sessions", session_id)
    pipe.expire(f"user:{user_id}:sessions", SESSION_TTL_SECONDS * 2)  # a bit longer

    pipe.execute()

    # 3. The session limit per user: if it's exceeded, remove the oldest one
    enforce_session_limit(user_id)

    return session_id


def enforce_session_limit(user_id: int):
    """If the user exceeds MAX_SESSIONS, remove the oldest one."""
    session_ids = r.smembers(f"user:{user_id}:sessions")

    if len(session_ids) <= MAX_SESSIONS_PER_USER:
        return

    # Find the oldest session
    oldest_session = None
    oldest_time = float('inf')

    for sid in session_ids:
        created_at = r.hget(f"session:{sid}", "created_at")
        if created_at:
            t = float(created_at)
            if t < oldest_time:
                oldest_time = t
                oldest_session = sid
        else:
            # The session already expired, clean up the set
            r.srem(f"user:{user_id}:sessions", sid)

    if oldest_session and len(session_ids) > MAX_SESSIONS_PER_USER:
        revoke_session(oldest_session)


def get_session(session_id: str) -> dict | None:
    """Gets a session's data. It renews the TTL (sliding)."""
    data = r.hgetall(f"session:{session_id}")
    if not data:
        return None

    # Renew the TTL (sliding)
    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


def revoke_session(session_id: str) -> bool:
    """Revokes a specific session."""
    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()

    return True


def revoke_all_sessions(user_id: int) -> int:
    """Revokes ALL of a user's sessions (a global logout)."""
    session_ids = r.smembers(f"user:{user_id}:sessions")
    if not session_ids:
        return 0

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

    return len(session_ids)


def list_user_sessions(user_id: int) -> list[dict]:
    """Lists all of a user's active sessions."""
    session_ids = r.smembers(f"user:{user_id}:sessions")
    sessions = []

    for sid in session_ids:
        data = r.hgetall(f"session:{sid}")
        if data:
            sessions.append({
                "session_id": sid,
                **data,
            })
        else:
            # Cleanup: the session expired but it's still in the set
            r.srem(f"user:{user_id}:sessions", sid)

    return sessions

Testing the session manager

"""
A quick test of the session manager.
"""
from sessions import (
    create_session,
    get_session,
    revoke_session,
    revoke_all_sessions,
    list_user_sessions,
)

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


# Clean up
for key in r.scan_iter("session:*"):
    r.delete(key)
for key in r.scan_iter("user:*:sessions"):
    r.delete(key)


print("=== Test 1: Creating a session ===")
sid_1 = create_session(user_id=42, metadata={
    "ip": "192.168.1.10",
    "user_agent": "Mozilla/5.0",
    "device_name": "MacBook Pro"
})
print(f"Session created: {sid_1[:16]}...")

print("\n=== Test 2: Getting the session ===")
data = get_session(sid_1)
print(f"Data: user_id={data['user_id']}, device={data['device_name']}")

print("\n=== Test 3: Multiple sessions for the same user ===")
sid_2 = create_session(user_id=42, metadata={"device_name": "iPhone"})
sid_3 = create_session(user_id=42, metadata={"device_name": "iPad"})
print(f"User 42 now has {len(list_user_sessions(42))} sessions")

print("\n=== Test 4: Listing the user's sessions ===")
for s in list_user_sessions(42):
    print(f"  - {s['session_id'][:16]}... ({s['device_name']})")

print("\n=== Test 5: Revoking a single session ===")
revoke_session(sid_2)
print(f"Sessions remaining: {len(list_user_sessions(42))}")

print("\n=== Test 6: Revoking ALL the sessions ===")
revoked = revoke_all_sessions(42)
print(f"Revoked: {revoked}")
print(f"Sessions remaining: {len(list_user_sessions(42))}")

Output:

=== Test 1: Creating a session ===
Session created: ABCdef123XYZ...

=== Test 2: Getting the session ===
Data: user_id=42, device=MacBook Pro

=== Test 3: Multiple sessions for the same user ===
User 42 now has 3 sessions

=== Test 4: Listing the user's sessions ===
  - ABCdef123XYZ... (MacBook Pro)
  - DEFghi456ABC... (iPhone)
  - GHIjkl789DEF... (iPad)

=== Test 5: Revoking a single session ===
Sessions remaining: 2

=== Test 6: Revoking ALL the sessions ===
Revoked: 2
Sessions remaining: 0

Integration with FastAPI + JWT

Create app.py:

"""
A FastAPI app with JWT + Redis sessions.
"""
import time
import jwt
from fastapi import FastAPI, HTTPException, Depends, Header, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel

from sessions import (
    create_session,
    get_session,
    revoke_session,
    revoke_all_sessions,
    list_user_sessions,
    SESSION_TTL_SECONDS,
)


JWT_SECRET = "change-me-in-production"
JWT_ALGORITHM = "HS256"
JWT_TTL_HOURS = 24


# A mock user database
fake_users_db = {
    "alice": {"user_id": 1, "username": "alice", "password": "secret"},
    "bob":   {"user_id": 2, "username": "bob",   "password": "secret"},
}


app = FastAPI(title="JWT + Sessions API")
security = HTTPBearer()


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


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


# ═══════════════════════════════════════════════════════════
# JWT helpers
# ═══════════════════════════════════════════════════════════


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:
    try:
        return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")


# ═══════════════════════════════════════════════════════════
# The auth dependency: verify the JWT + the session
# ═══════════════════════════════════════════════════════════


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """
    Verifies the JWT + verifies that the session exists.
    If the session was revoked, it returns a 401.
    """
    token = credentials.credentials
    payload = decode_jwt(token)

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

    # ✨ Here's the magic: verify the session exists in Redis
    session = get_session(session_id)
    if not session:
        raise HTTPException(401, "Session revoked or expired. Please login again.")

    # Return the combined context
    return {
        "user_id": int(session["user_id"]),
        "session_id": session_id,
        "device_name": session.get("device_name", ""),
    }


# ═══════════════════════════════════════════════════════════
# Endpoints
# ═══════════════════════════════════════════════════════════


@app.post("/auth/login", response_model=LoginResponse)
def login(req: LoginRequest, request: Request):
    """Login: creates a session in Redis + returns a JWT."""
    user = fake_users_db.get(req.username)
    if not user or user["password"] != req.password:
        raise HTTPException(401, "Invalid credentials")

    user_id = user["user_id"]

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

    # Create the JWT with the session_id included
    token = create_jwt(user_id, session_id)

    return LoginResponse(
        access_token=token,
        session_id=session_id,
        user_id=user_id,
    )


@app.get("/me")
def me(user: dict = Depends(get_current_user)):
    """A protected endpoint: it requires a JWT + a valid session."""
    return {
        "user_id": user["user_id"],
        "current_session": user["session_id"],
        "device": user["device_name"],
    }


@app.post("/auth/logout")
def logout(user: dict = Depends(get_current_user)):
    """A local logout: it revokes the current session."""
    revoke_session(user["session_id"])
    return {"message": "Logged out from this device"}


@app.post("/auth/logout-all")
def logout_all(user: dict = Depends(get_current_user)):
    """A global logout: it revokes ALL of the user's sessions."""
    revoked = revoke_all_sessions(user["user_id"])
    return {"message": f"Logged out from {revoked} devices"}


@app.get("/auth/sessions")
def my_sessions(user: dict = Depends(get_current_user)):
    """Lists all of the user's active sessions."""
    sessions = list_user_sessions(user["user_id"])
    # Mark the current session
    for s in sessions:
        s["is_current"] = (s["session_id"] == user["session_id"])
    return {"sessions": sessions, "total": len(sessions)}


@app.post("/auth/sessions/{session_id}/revoke")
def revoke_specific_session(session_id: str, user: dict = Depends(get_current_user)):
    """Revokes a specific session of the user's (not the current one)."""
    if session_id == user["session_id"]:
        raise HTTPException(400, "Use /auth/logout to revoke current session")

    # Verify that the session belongs to the current user
    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"}

Testing the API

uvicorn app:app --reload &
sleep 2

# 1. Login (it creates session 1)
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"}' | python -m json.tool

# Expected output:
# {
#   "access_token": "eyJhbGc...",
#   "session_id": "ABC123...",
#   "user_id": 1
# }

# Save the token
TOKEN_1=$(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"}' | python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 2. Log in from another "device" (it creates session 2)
TOKEN_2=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -H "X-Device-Name: iPhone" \
  -d '{"username": "alice", "password": "secret"}' | python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 3. Access /me with token 1
curl -s -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/me

# 4. List my active sessions (it should show 2)
curl -s -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/auth/sessions | python -m json.tool

# 5. A global logout from device 1
curl -s -X POST -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/auth/logout-all

# 6. Try to access with token 2 (it should fail — the session was revoked)
curl -s -H "Authorization: Bearer $TOKEN_2" http://localhost:8000/me
# {"detail": "Session revoked or expired. Please login again."}

The critical part of step 6: TOKEN_2 (the JWT) has technically NOT expired — its exp claim is still valid 24 hours into the future. But the server returns a 401 because the session was revoked in Redis. That's what JWT alone CAN'T do.

A sliding TTL for active sessions

When a user is actively using the app, their session should stay alive. When they stop using it for a long time, it should expire automatically. That's a sliding TTL.

In the implementation you saw, get_session() renews the TTL on every access:

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

    # Renew the TTL on every access
    r.expire(f"session:{session_id}", SESSION_TTL_SECONDS)
    r.hset(f"session:{session_id}", "last_accessed", str(time.time()))

    return data

The resulting behavior:

• The user uses the app every 5 min → the TTL always renews → the session lives indefinitely
• The user abandons the app → there are no accesses → the TTL counts down → it expires after 30 min of inactivity

This is exactly the classic UX: "we keep you logged in while you use the app, we log you out for inactivity."

Sliding vs absolute timeout

Sometimes you need both:

SESSION_SLIDING_TTL = 30 * 60       # 30 min of inactivity
SESSION_ABSOLUTE_TTL = 24 * 3600    # 24 hours maximum


def create_session_with_absolute_ttl(user_id):
    session_id = secrets.token_urlsafe(24)
    now = time.time()

    r.hset(f"session:{session_id}", mapping={
        "user_id": str(user_id),
        "created_at": str(now),
        "absolute_expires_at": str(now + SESSION_ABSOLUTE_TTL),
    })
    r.expire(f"session:{session_id}", SESSION_SLIDING_TTL)

    return session_id


def get_session_with_absolute_check(session_id):
    data = r.hgetall(f"session:{session_id}")
    if not data:
        return None

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

    # Renew the sliding TTL
    r.expire(f"session:{session_id}", SESSION_SLIDING_TTL)

    return data

The behavior: the session renews with every use (sliding), but it NEVER lasts more than 24 hours total (absolute). A classic pattern for banking apps.

Local logout vs global logout

When a user changes their password

@app.post("/auth/change-password")
def change_password(
    new_password: str,
    user: dict = Depends(get_current_user)
):
    # 1. Update the DB with the new password hash
    db.update_user_password(user["user_id"], new_password)

    # 2. Global logout: revoke every session (including the current one)
    revoke_all_sessions(user["user_id"])

    # 3. Create a new session so the current user doesn't get logged out
    new_session_id = create_session(user["user_id"])
    new_token = create_jwt(user["user_id"], new_session_id)

    return {
        "message": "Password updated. All other sessions logged out.",
        "new_access_token": new_token,
    }

The flow: you change your password → every existing session dies (including the current request's) → we create a new session so YOU don't get logged out → the client updates its token.

Active sessions: the user's devices

A common feature: "see my connected devices." Like Gmail or WhatsApp:

Your account is active on:

🖥  MacBook Pro · Buenos Aires · 5 min ago  ← The current session
📱 iPhone · Madrid · 2 hours ago
💻 Linux Workstation · São Paulo · 5 days ago

[Log out on all devices]

The implementation:

@app.get("/auth/sessions")
def list_my_sessions(user: dict = Depends(get_current_user)):
    sessions = list_user_sessions(user["user_id"])

    # Enrich it with info that's useful for the UI
    enriched = []
    for s in sessions:
        last_accessed = float(s.get("last_accessed", 0))
        age_seconds = time.time() - last_accessed

        if age_seconds < 60:
            last_active = "Just now"
        elif age_seconds < 3600:
            last_active = f"{int(age_seconds / 60)} min ago"
        elif age_seconds < 86400:
            last_active = f"{int(age_seconds / 3600)} hours ago"
        else:
            last_active = f"{int(age_seconds / 86400)} days ago"

        enriched.append({
            "session_id": s["session_id"],
            "device_name": s.get("device_name", "Unknown"),
            "ip": s.get("ip", ""),
            "last_active": last_active,
            "is_current": s["session_id"] == user["session_id"],
        })

    return {"sessions": enriched}

Security: best practices

1. Generating the session_id

✅ Right:

import secrets
session_id = secrets.token_urlsafe(24)  # 24 bytes = 192 bits of entropy

❌ Wrong:

import random
session_id = str(random.randint(0, 1_000_000))  # easily guessable

secrets.token_urlsafe() uses a cryptographically secure source. Enough entropy to prevent brute-forcing.

2. Don't include sensitive data in the JWT claims

JWT claims are visible to the client (base64-encoded, not encrypted).

✅ Right:

{"user_id": 42, "session_id": "abc123"}

❌ Wrong:

{"user_id": 42, "password_hash": "...", "credit_card": "..."}

Sensitive data is stored on the server (Redis or the DB), NOT in the JWT.

3. Roles from Redis, not from the JWT

❌ Wrong (roles in the JWT):

# The JWT includes a "roles": ["admin"] claim
# If you revoke the admin role, the old JWT STILL says "admin" until it expires

✅ Right (roles in Redis):

# The JWT only has the user_id
# On every request: r.hget(f"user:{user_id}:roles", ...)
# If you revoke the admin role, the next request sees it immediately

Any data that can change and needs immediate revocation → in Redis, not in the JWT.

4. HTTPS is mandatory

Sessions over HTTP expose the JWT on every request. Any MITM can steal it.

# In production:
# - Force HTTPS (redirect HTTP → HTTPS)
# - Set the cookie with the Secure flag
# - An HSTS header

5. Rate limit the login endpoint

To prevent brute force:

# 5 failed attempts per IP per minute
login_limiter = SlidingWindowLimiter(limit=5, window=60)


@app.post("/auth/login")
def login(req: LoginRequest, request: Request):
    ip = request.client.host
    allowed, _, _ = login_limiter.check(f"login:{ip}")
    if not allowed:
        raise HTTPException(429, "Too many login attempts. Try again later.")

    # ... the rest of the flow

6. Logging auth events

To detect attacks and for auditing:

import logging
logger = logging.getLogger("auth")


@app.post("/auth/login")
def login(req: LoginRequest, request: Request):
    user = fake_users_db.get(req.username)

    if not user or user["password"] != req.password:
        logger.warning(f"Failed login: user={req.username}, ip={request.client.host}")
        raise HTTPException(401, "Invalid credentials")

    logger.info(f"Successful login: user_id={user['user_id']}, ip={request.client.host}")
    # ...

Troubleshooting

Problem 1: Sessions disappear even though the user is active

Cause: The TTL isn't being renewed. The r.expire() inside get_session() is probably being skipped.

Solution: Check that every path that validates the session calls get_session() (which renews the TTL). If you have auth middleware, it should use get_session().

Problem 2: The user sees their own sessions from months ago

Cause: The user:{id}:sessions set isn't cleaned up when the individual sessions expire.

Solution: In list_user_sessions(), verify that each session_id has its data, and clean up the set if it doesn't:

def list_user_sessions(user_id):
    session_ids = r.smembers(f"user:{user_id}:sessions")
    sessions = []

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

    return sessions

Problem 3: A race condition when creating a session

Cause: Two workers create sessions simultaneously, and enforce_session_limit() sees different counts.

Solution: Use a Lua script for atomicity, or accept that occasionally there could be 6 sessions instead of 5 (it isn't critical).

Problem 4: The JWT doesn't include a session_id

Cause: The client is using an old JWT (from before you implemented sessions).

Solution: Force a re-login. You can detect this:

session_id = payload.get("session_id")
if not session_id:
    raise HTTPException(401, "Token outdated. Please login again.")

Problem 5: Redis's memory grows with abandoned sessions

Cause: Users who never log out leave their sessions behind until they expire naturally.

Solution: A periodic cleanup of orphaned keys (similar to capsule 03):

def cleanup_orphan_user_session_sets():
    """Cleans up user:{id}:sessions sets that have no real session keys."""
    for key in r.scan_iter("user:*:sessions"):
        session_ids = r.smembers(key)
        for sid in session_ids:
            if not r.exists(f"session:{sid}"):
                r.srem(key, sid)
        if r.scard(key) == 0:
            r.delete(key)

Run it every night.

Problem 6: Revoked sessions reappear after the cleanup

Cause: There's a loop recreating sessions automatically (a bad auto-refresh pattern).

Solution: Make sure revoke_session() actually deletes the session, and that no endpoint recreates it silently. Audit with MONITOR if you're suspicious:

redis-cli MONITOR | grep "session:abc123"

Exercises

Exercise 1: Creating and revoking a session (Easy)

Create a session for user_id=42. Read it. Revoke it. Try to read it again (it should return None).

from sessions import create_session, get_session, revoke_session


# Create
sid = create_session(user_id=42)
print(f"Session created: {sid}")

# Read
data = get_session(sid)
print(f"Data: {data}")

# Revoke
revoke_session(sid)
print("Revoked")

# Try to read it again
data_after = get_session(sid)
print(f"After the revoke: {data_after}")  # None

Session created: x9z8y7w6v5u4t3s2r1q0...
Data: {'user_id': '42', 'created_at': '1714069200.5', ...}
Revoked
After the revoke: None

Exercise 2: A global logout (Easy-Medium)

Create 3 sessions for the same user. List the sessions (there should be 3). Revoke them all with revoke_all_sessions. Verify there are no sessions left.

from sessions import (
    create_session,
    list_user_sessions,
    revoke_all_sessions,
)


user_id = 42

# A prior cleanup
revoke_all_sessions(user_id)

# Create 3 sessions
sid1 = create_session(user_id, {"device_name": "MacBook"})
sid2 = create_session(user_id, {"device_name": "iPhone"})
sid3 = create_session(user_id, {"device_name": "iPad"})

# List them
sessions = list_user_sessions(user_id)
print(f"Active sessions: {len(sessions)}")
for s in sessions:
    print(f"  - {s['device_name']}")

# Revoke all
revoked = revoke_all_sessions(user_id)
print(f"\nRevoked: {revoked}")

# Verify
print(f"Active sessions now: {len(list_user_sessions(user_id))}")

Active sessions: 3
  - MacBook
  - iPhone
  - iPad

Revoked: 3
Active sessions now: 0

Exercise 3: A sliding TTL in action (Medium)

Create a session with a 10-second TTL. Read the session every 3 seconds (5 times). Verify the TTL never drops below ~10 (it renews on every read). Then wait 12 sec without reading and verify it expired.

import time
from sessions import create_session, get_session, SESSION_TTL_SECONDS
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


# Change the TTL for the test (in production it's 30 min)
import sessions
sessions.SESSION_TTL_SECONDS = 10  # 10 sec for the test

sid = create_session(user_id=42)
print(f"Session created with TTL=10s")
print(f"Initial TTL: {r.ttl(f'session:{sid}')}s")

# Read 5 times, every 3 sec
for i in range(5):
    time.sleep(3)
    data = get_session(sid)
    ttl = r.ttl(f"session:{sid}")
    print(f"Read {i+1}: data={'OK' if data else 'NULL'}, TTL={ttl}s")

# Wait 12 sec with no reads
print("\nWaiting 12 sec without accessing it...")
time.sleep(12)

# Verify it expired
data = get_session(sid)
print(f"After 12s: data={'OK' if data else 'NULL (it expired)'}")

Session created with TTL=10s
Initial TTL: 10s
Read 1: data=OK, TTL=10s   # renewed
Read 2: data=OK, TTL=10s   # renewed
Read 3: data=OK, TTL=10s
Read 4: data=OK, TTL=10s
Read 5: data=OK, TTL=10s

Waiting 12 sec without accessing it...
After 12s: data=NULL (it expired)

Exercise 4: Login with FastAPI + curl (Medium)

Implement the capsule's app. Log in with curl. Use the returned token to access /me. Then log out and verify that /me no longer works.

# Assuming you have the app running (uvicorn app:app --reload)

# 1. Login
RESPONSE=$(curl -s -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "password": "secret"}')
echo "Login: $RESPONSE"

TOKEN=$(echo $RESPONSE | python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 2. /me (it should work)
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/me
# {"user_id": 1, ...}

# 3. Logout
curl -s -X POST -H "Authorization: Bearer $TOKEN" http://localhost:8000/auth/logout

# 4. /me (it should fail — the session was revoked)
curl -s -H "Authorization: Bearer $TOKEN" http://localhost:8000/me
# {"detail": "Session revoked or expired. Please login again."}

Exercise 5: Listing sessions and revoking a specific one (Medium)

Implement the "see my devices and disconnect one" flow: log in with 3 different device_names, list the sessions, revoke a specific one (not the current one), and verify that 2 remain.

# Log in 3 times with different devices
TOKEN_1=$(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)

TOKEN_2=$(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)

TOKEN_3=$(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 the sessions from TOKEN_1
curl -s -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/auth/sessions | jq

# Output: 3 sessions with is_current=true on one of them

# Get the iPhone's session_id
IPHONE_SID=$(curl -s -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/auth/sessions | \
  jq -r '.sessions[] | select(.device_name == "iPhone") | .session_id')

# Revoke the iPhone from the MacBook (TOKEN_1)
curl -s -X POST -H "Authorization: Bearer $TOKEN_1" \
  http://localhost:8000/auth/sessions/$IPHONE_SID/revoke

# Verify: TOKEN_2 (the iPhone) no longer works
curl -s -H "Authorization: Bearer $TOKEN_2" http://localhost:8000/me
# {"detail": "Session revoked or expired."}

# TOKEN_1 and TOKEN_3 keep working
curl -s -H "Authorization: Bearer $TOKEN_1" http://localhost:8000/me
curl -s -H "Authorization: Bearer $TOKEN_3" http://localhost:8000/me

Exercise 6: A password change with a global logout (Hard)

Implement the POST /auth/change-password endpoint, which: updates the password (mock), revokes all the user's sessions, creates a new session for the current request, and returns a new JWT. The current user should NOT be logged out, but every other device should be.

class ChangePasswordRequest(BaseModel):
    new_password: str


@app.post("/auth/change-password")
def change_password(
    req: ChangePasswordRequest,
    user: dict = Depends(get_current_user)
):
    user_id = user["user_id"]

    # 1. Update the password (a mock — in prod, hash + DB)
    fake_users_db_by_id[user_id]["password"] = req.new_password

    # 2. Revoke every session (including the current one!)
    revoked = revoke_all_sessions(user_id)

    # 3. Create a new session for the current device (so we don't log out the current user)
    new_session_id = create_session(user_id, metadata={
        "device_name": user.get("device_name", "Unknown"),
    })
    new_token = create_jwt(user_id, new_session_id)

    return {
        "message": f"Password updated. {revoked} sessions revoked.",
        "new_access_token": new_token,
        "old_token_invalid": True,
    }

TOKEN_OLD=$(curl -s -X POST http://localhost:8000/auth/login ... | jq -r .access_token)

# Change the password
RESPONSE=$(curl -s -X POST http://localhost:8000/auth/change-password \
  -H "Authorization: Bearer $TOKEN_OLD" \
  -H "Content-Type: application/json" \
  -d '{"new_password": "newpass123"}')

TOKEN_NEW=$(echo $RESPONSE | jq -r .new_access_token)

# The old token does NOT work (the old session was revoked)
curl -s -H "Authorization: Bearer $TOKEN_OLD" http://localhost:8000/me
# {"detail": "Session revoked..."}

# The new token DOES work
curl -s -H "Authorization: Bearer $TOKEN_NEW" http://localhost:8000/me
# {"user_id": 1, ...}

Summary

In this capsule you learned:

• JWT alone doesn't allow immediate revocation — a compromised token stays valid until it expires naturally
• The professional pattern: JWT + a Redis session — the JWT verifies identity (stateless), Redis holds the revocable mutable state
• The implementation: HSET for the session's data, a set of session_ids per user, a sliding TTL for active sessions
• The login flow: validate the credentials → create the session in Redis → issue a JWT with the session_id in its claims
• The auth dependency: verify the JWT (no DB) + verify the session exists in Redis (1 ms) = <2 ms total
• Local logout vs global logout: revoke 1 session vs revoke all of the user's
• Listing active sessions: showing the user their connected devices (standard UX)
• Sliding vs absolute TTL: keep the session alive with sliding, but expire it absolutely after N hours
• Security best practices: secrets.token_urlsafe() for the session_id, don't include sensitive data in the JWT, roles from Redis not the JWT, rate limiting on login, event logging
• A password change = a global logout + a new session for the current device

The JWT + Redis sessions pattern is what separates a simple API from a production-ready one. Every API you know (GitHub, Slack, Notion, Stripe) uses it.

Additional resources

1. JWT.io — A decoder and info about the JWT structure
2. PyJWT Docs — The Python client you used in this capsule
3. OWASP Session Management Cheat Sheet — Security best practices
4. Stop using JWT for sessions — The classic critique of using JWT on its own
5. The Definitive Guide to JWT vs Sessions — A modern comparison that recommends combining them
6. Auth0: Token Best Practices — Production patterns

What's next?

In Capsule 05 you consolidate the whole module in the Rate Limiter Service mini-project — a complete API with:

• Multi-tier sliding window rate limiting (free/pro/enterprise)
• JWT + Redis sessions with every flow (login, logout, logout-all, list-sessions)
• Load tests simulating 1000 concurrent requests
• Correct standard HTTP headers
• Automatic cleanup of orphaned sessions

It's module 3's final capsule. After that, only M4 (Pub/Sub & FastAPI) and M5 (the capstone project) remain.

Keep your workspace and Redis running. Let's go.