A Practical Guide to Multi-Level Caching with Redis

[Shrinit]

"What if you could cut your database load by 90% without adding more servers?”

In modern web applications, caching is crucial for performance optimization. This blog explores implementing a robust multi-level caching strategy using Redis as the second layer, demonstrated through a real-world E-Learning platform.

Table of Contents

1. Understanding Multi-Level Caching
2. Architecture Overview
3. Implementation Deep Dive
4. Authentication Caching
5. Course Data Caching
6. Cache Invalidation Strategies
7. Performance Monitoring
8. Production Considerations

Understanding Multi-Level Caching

Multi-level caching implements multiple cache layers with different characteristics:

• L1 Cache (Memory): Ultra-fast in-memory cache for hot data
• L2 Cache (Redis): Distributed cache shared across application instances
• L3 (Database): Persistent storage as the source of truth

Benefits & Performance Gains

• Reduced Latency: Memory access is ~100x faster than Redis
• Scalability: Redis enables horizontal scaling
• Fault Tolerance: Multiple fallback layers
• Cost Efficiency: Reduces database load significantly

Speed Comparison

Cache Layer	Typical Latency	Throughput	Use Case
L1 (Memory)	0.1-1ms	100K+ ops/sec	Hot data, session info
L2 (Redis)	1-5ms	50K ops/sec	Distributed cache
L3 (Database)	10-100ms	1K ops/sec	Source of truth

Real Performance Gains:

• Course retrieval: 50ms → 0.5ms (100x faster)
• Course listings: 200ms → 2ms (100x faster)
• Auth validation: 25ms → 0.2ms (125x faster)

Architecture Overview

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   L1 (Memory)   │───▶│   L2 (Redis)    │───▶│  L3 (MongoDB)   │
│   TTL: 5 min    │    │   TTL: 5 min    │    │   Source of     │
│   Ultra Fast    │    │   Distributed   │    │     Truth       │
└─────────────────┘    └─────────────────┘    └─────────────────┘

Implementation Deep Dive

1. Memory Cache Layer (L1)

First, let’s implement the in-memory cache with TTL support:

# services/memory_cache.py
import asyncio
import time
from typing import Any, Optional, Dict

class AsyncInMemoryCache:
    def __init__(self):
        self._store: Dict[str, tuple[float, Any]] = {}
        self._locks: Dict[str, asyncio.Lock] = {}
        self._global_lock = asyncio.Lock()

    def _now(self) -> float:
        return time.monotonic()

    async def get(self, key: str) -> Optional[Any]:
        item = self._store.get(key)
        if not item:
            return None
        expires_at, value = item
        if expires_at != 0 and self._now() > expires_at:
            # expired
            self._store.pop(key, None)
            return None
        return value

    async def set(self, key: str, value: Any, ttl: int = 0) -> None:
        expires_at = self._now() + ttl if ttl and ttl > 0 else 0
        self._store[key] = (expires_at, value)

    async def delete(self, key: str) -> None:
        self._store.pop(key, None)

    async def pattern_delete(self, prefix: str) -> None:
        # simple prefix match to clear many keys
        keys = [k for k in self._store.keys() if k.startswith(prefix)]
        for k in keys:
            self._store.pop(k, None)

    async def get_lock(self, key: str) -> asyncio.Lock:
        # per-key lock for dogpile protection
        async with self._global_lock:
            if key not in self._locks:
                self._locks[key] = asyncio.Lock()
            return self._locks[key]

# Singleton instance
memory_cache = AsyncInMemoryCache()

2. Cache Key Management

Organize cache keys systematically:

# services/cache_keys.py
def course_key(course_id: str) -> str:
    return f"course:{course_id}"

def courses_list_key(filters_hash: str) -> str:
    return f"courses_list:{filters_hash}"

def user_session_key(user_id: str) -> str:
    return f"user_session:{user_id}"

def refresh_tokens_key(user_id: str) -> str:
    return f"refresh_tokens:{user_id}"

def blacklisted_jti_key(jti: str) -> str:
    return f"blacklisted_tokens:{jti}"

def analytics_course_key(course_id: str) -> str:
    return f"analytics:course:{course_id}"

3. Configuration Setup

Configure your application dependencies:

# config.py
from pydantic_settings import BaseSettings
from pydantic import Field, validator

class Settings(BaseSettings):
    MONGO_URI: str = Field(..., description="MongoDB connection URI")
    REDIS_URL: str = Field(..., description="Redis connection URL")
    JWT_SECRET: str = Field(..., min_length=32, description="JWT secret key")
    
    ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(default=15, ge=1, le=1440)
    REFRESH_TOKEN_EXPIRE_DAYS: int = Field(default=7, ge=1, le=30)
    
    @validator('REDIS_URL')
    def validate_redis_url(cls, v):
        if not v.startswith('redis://'):
            raise ValueError('REDIS_URL must be a valid Redis connection string')
        return v

    class Config:
        env_file = ".env"

settings = Settings()

# deps.py
from pymongo.database import Database
from pymongo import MongoClient
import redis.asyncio as aioredis
from fastapi import Request
from redis.asyncio import Redis

def create_mongo_client(uri: str) -> MongoClient:
    return MongoClient(uri, maxPoolSize=100, serverSelectionTimeoutMS=5000)

def create_redis_client(url: str):
    return aioredis.from_url(url, encoding="utf-8", decode_responses=True)

def get_db(request: Request) -> Database:
    return request.app.state.db

def get_redis(request: Request) -> Redis:
    return request.app.state.redis

Authentication Caching

Authentication is a perfect use case for multi-level caching due to high frequency and read-heavy patterns.

JWT Token Management

# routers/user_auth/auth.py
from fastapi import APIRouter, Depends, HTTPException
from redis.asyncio import Redis
from pymongo.database import Database
import json
from datetime import datetime, timezone

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

SESSION_TTL = 60 * 60 * 24      # 24 hours
REFRESH_TTL = 60 * 60 * 24 * 7  # 7 days

@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Database = Depends(get_db),
    r: Redis = Depends(get_redis)
):
    # Authenticate user
    user = users.get_user_by_email(db, form_data.username)
    if not user or not users.verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    user_id = str(user["_id"])
    access_token = create_access_token({"sub": user_id, "role": user["role"]})
    refresh_token = create_refresh_token({"sub": user_id, "role": user["role"]})

    # Cache user session in Redis
    await r.set(
        user_session_key(user_id), 
        json.dumps({"email": user["email"], "role": user["role"]}), 
        ex=SESSION_TTL
    )
    await r.set(refresh_tokens_key(user_id), refresh_token, ex=REFRESH_TTL)

    return {"access_token": access_token, "refresh_token": refresh_token}

@router.delete("/logout")
async def logout(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/login")), 
    r: Redis = Depends(get_redis)
):
    decoded = decode_token(token)
    jti = decoded["jti"]
    ttl = decoded["exp"] - int(datetime.now(timezone.utc).timestamp())
    
    # Blacklist token
    await r.set(blacklisted_jti_key(jti), "true", ex=ttl)
    
    # Clear user session
    await r.delete(user_session_key(decoded['sub']))
    await r.delete(refresh_tokens_key(decoded['sub']))

    return {"message": "Logged out successfully"}

Token Validation with Caching

# auth/dependencies.py
from fastapi import Depends, HTTPException
from redis.asyncio import Redis

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    r: Redis = Depends(get_redis)
):
    try:
        payload = decode_token(token)
        jti = payload["jti"]
        
        # Check if token is blacklisted (Redis only - no need for L1)
        if await r.get(blacklisted_jti_key(jti)):
            raise HTTPException(status_code=401, detail="Token has been revoked")
        
        user_id = payload["sub"]
        
        # Try to get user session from cache
        cached_session = await r.get(user_session_key(user_id))
        if cached_session:
            session_data = json.loads(cached_session)
            return {
                "_id": user_id,
                "email": session_data["email"],
                "role": session_data["role"]
            }
        
        # Fallback to database if not in cache
        user = await get_user_from_db(user_id)
        if not user:
            raise HTTPException(status_code=401, detail="User not found")
            
        return user
        
    except Exception as e:
        raise HTTPException(status_code=401, detail="Invalid token")

Course Data Caching

Course data represents the core of our caching strategy with complex invalidation requirements.

Multi-Level Course Service

# services/course_service.py
import json
import hashlib
from typing import Dict, Any, Optional
from redis.asyncio import Redis
from pymongo.database import Database
from fastapi.concurrency import run_in_threadpool

# TTL Configuration
COURSE_TTL = 60 * 5          # 5 minutes for individual courses
COURSE_LIST_TTL = 60 * 2     # 2 minutes for course lists

def _filters_key(q: Optional[str], filters: Dict[str, Any], page: int, page_size: int, sort_by: str) -> str:
    """Generate unique cache key for course list filters"""
    payload = {"q": q, "filters": filters, "page": page, "page_size": page_size, "sort_by": sort_by}
    digest = hashlib.sha1(json.dumps(payload, sort_keys=True).encode()).hexdigest()
    return digest

async def get_course(db: Database, r: Redis, course_id: str) -> Optional[Dict[str, Any]]:
    """Retrieve course using two-level caching"""
    key = course_key(course_id)

    # Try L1 cache first
    cached = await memory_cache.get(key)
    if cached:
        await hit(r, "courses")  # Track cache hit
        return cached

    # Use lock to prevent cache stampede
    lock = await memory_cache.get_lock(key)
    async with lock:
        # Double-check L1 cache
        cached_again = await memory_cache.get(key)
        if cached_again:
            await hit(r, "courses")
            return cached_again

        # Try L2 (Redis) cache
        cached_l2 = await r.get(key)
        if cached_l2:
            payload = json.loads(cached_l2)
            await memory_cache.set(key, payload, ttl=COURSE_TTL)
            await hit(r, "courses")
            return payload

        # Cache miss - fetch from database
        await miss(r, "courses")  # Track cache miss
        doc = await run_in_threadpool(repo.get_course_by_id, db, course_id)
        if doc:
            # Store in both cache levels
            await r.set(key, json.dumps(doc, cls=JSONEncoder), ex=COURSE_TTL)
            await memory_cache.set(key, doc, ttl=COURSE_TTL)
        return doc

async def list_courses(
    db: Database, r: Redis, *, 
    q: Optional[str], filters: Dict[str, Any], 
    page: int, page_size: int, sort_by: str
):
    """List courses with filtering using two-level caching"""
    digest = _filters_key(q, filters, page, page_size, sort_by)
    key = courses_list_key(digest)

    # Try L1 cache first
    cached = await memory_cache.get(key)
    if cached:
        await hit(r, "courses_list")
        return cached

    # Use lock to prevent cache stampede
    lock = await memory_cache.get_lock(key)
    async with lock:
        # Double-check L1 cache
        cached_again = await memory_cache.get(key)
        if cached_again:
            await hit(r, "courses_list")
            return cached_again

        # Try L2 (Redis) cache
        cached_l2 = await r.get(key)
        if cached_l2:
            payload = json.loads(cached_l2)
            await memory_cache.set(key, payload, ttl=COURSE_LIST_TTL)
            await hit(r, "courses_list")
            return payload

        # Cache miss - fetch from database
        await miss(r, "courses_list")
        total, items = await run_in_threadpool(
            repo.list_courses, db, q=q, filters=filters, 
            page=page, page_size=page_size, sort_by=sort_by
        )
        payload = {"total": total, "page": page, "page_size": page_size, "items": items}
        
        # Store in both cache levels
        await r.set(key, json.dumps(payload, cls=JSONEncoder), ex=COURSE_LIST_TTL)
        await memory_cache.set(key, payload, ttl=COURSE_LIST_TTL)
        return payload

Course API Endpoints

# routers/courses_route/courses.py
from fastapi import APIRouter, Depends, Query
from typing import Optional, List

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

@router.get("")
async def list_courses(
    search: Optional[str] = Query(None, description="Full-text search"),
    category: Optional[str] = None,
    difficulty: Optional[str] = Query(None, regex="^(beginner|intermediate|advanced)$"),
    sort_by: str = Query("recent", regex="^(recent|popular|top_rated|duration)$"),
    page: int = Query(1, ge=1),
    page_size: int = Query(12, ge=1, le=100),
    db: Database = Depends(get_db),
    r: Redis = Depends(get_redis),
):
    filters = {
        **({k: v} for k, v in {
            "category": category,
            "difficulty": difficulty
        }.items() if v is not None)
    }
    
    result = await course_service.list_courses(
        db, r, q=search, filters=filters, 
        page=page, page_size=page_size, sort_by=sort_by
    )
    return result

@router.get("/{course_id}")
async def get_course(
    course_id: str, 
    db: Database = Depends(get_db), 
    r: Redis = Depends(get_redis)
):
    doc = await course_service.get_course(db, r, course_id)
    if not doc:
        raise HTTPException(status_code=404, detail="Course not found")
    return doc

@router.post("", status_code=201)
async def create_course(
    payload: CourseCreate, 
    db: Database = Depends(get_db), 
    r: Redis = Depends(get_redis),
    user=Depends(get_current_user)
):
    doc = await course_service.create_course(db, r, payload.dict())
    return doc

Cache Invalidation Strategies

Proper cache invalidation is crucial for data consistency.

Smart Invalidation Service

# services/cache_service.py
import asyncio
from typing import Dict, Any
from redis.asyncio import Redis
from bson import ObjectId

async def invalidate_course_cache(r: Redis, course_id: str) -> Dict[str, Any]:
    """Invalidate all caches related to a specific course"""
    try:
        # Validate course_id to prevent injection
        ObjectId(course_id)
        
        key = course_key(course_id)
        analytics_key = analytics_course_key(course_id)

        # Parallel deletion of related keys
        await asyncio.gather(
            memory_cache.delete(key),
            memory_cache.delete(analytics_key),
            r.delete(key, analytics_key),
            memory_cache.pattern_delete("courses_list:"),
            _delete_redis_pattern(r, "courses_list:*")
        )

        return {"message": f"Cache cleared for course {course_id}"}
        
    except Exception as e:
        raise ValueError(f"Invalid course_id: {str(e)}")

async def _delete_redis_pattern(r: Redis, pattern: str):
    """Efficiently delete Redis keys by pattern"""
    cursor = 0
    deleted_count = 0
    
    while True:
        cursor, keys = await r.scan(cursor=cursor, match=pattern, count=500)
        if keys:
            # Use pipeline for batch deletion
            async with r.pipeline() as pipe:
                for key in keys:
                    pipe.delete(key)
                await pipe.execute()
                deleted_count += len(keys)
        if cursor == 0:
            break

Update Operations with Cache Management

async def create_course(db: Database, r: Redis, data: Dict[str, Any]) -> Dict[str, Any]:
    """Create course and manage cache invalidation"""
    # Insert new course
    doc = await run_in_threadpool(repo.insert_course, db, data)
    
    # Invalidate course lists since they're now outdated
    await _invalidate_course_lists(r)
    
    # Cache the new course immediately
    key = course_key(doc["_id"])
    await r.set(key, json.dumps(doc, cls=JSONEncoder), ex=COURSE_TTL)
    await memory_cache.set(key, doc, ttl=COURSE_TTL)
    
    return doc

async def update_course_module(db: Database, r: Redis, course_id: str, module_id: str, patch: Dict[str, Any]):
    """Update module and invalidate related caches"""
    doc = await run_in_threadpool(repo.update_module, db, course_id, module_id, patch)
    if doc:
        # Invalidate course caches
        key = course_key(course_id)
        await memory_cache.delete(key)
        await r.delete(key)
        await _invalidate_course_lists(r)
        
        # Cache updated course
        await r.set(key, json.dumps(doc, cls=JSONEncoder), ex=COURSE_TTL)
        await memory_cache.set(key, doc, ttl=COURSE_TTL)
    return doc

async def _invalidate_course_lists(r: Redis) -> None:
    """Invalidate all cached course lists"""
    # Clear from both cache levels
    await memory_cache.pattern_delete("courses_list:")
    await _delete_redis_pattern(r, "courses_list:*")

Performance Monitoring

Track cache performance to optimize your strategy:

# services/cache_stats.py
from redis.asyncio import Redis

async def hit(r: Redis, cache_type: str):
    """Record cache hit"""
    await r.incr(f"cache_hits:{cache_type}")

async def miss(r: Redis, cache_type: str):
    """Record cache miss"""
    await r.incr(f"cache_misses:{cache_type}")

async def get_cache_stats(r: Redis) -> Dict[str, Any]:
    """Get comprehensive cache statistics"""
    try:
        # L1 stats
        l1_size = len(getattr(memory_cache, '_store', {}))

        # L2 stats from Redis
        info = await r.info()
        redis_keys = info.get("db0", {}).get("keys", 0) if "db0" in info else 0
        memory_used = info.get("used_memory_human", "N/A")

        return {
            "memory_cache_size": l1_size,
            "redis_keys": redis_keys,
            "redis_memory_used": memory_used,
            "redis_hits": info.get("keyspace_hits", 0),
            "redis_misses": info.get("keyspace_misses", 0),
        }
        
    except Exception as e:
        return {"error": "Failed to retrieve cache stats"}

Cache Monitoring Endpoint

# routers/cache_route/cache.py
@router.get("/stats")
async def cache_statistics(r: Redis = Depends(get_redis)):
    """Get cache performance statistics"""
    stats = await get_cache_stats(r)
    
    # Calculate hit ratios
    total_hits = stats.get("redis_hits", 0)
    total_misses = stats.get("redis_misses", 0)
    total_requests = total_hits + total_misses
    
    if total_requests > 0:
        stats["hit_ratio"] = round((total_hits / total_requests) * 100, 2)
    else:
        stats["hit_ratio"] = 0
    
    return stats

@router.delete("/invalidate/{course_id}")
async def invalidate_course(
    course_id: str,
    r: Redis = Depends(get_redis),
    user=Depends(require_role("admin"))
):
    """Manually invalidate course cache"""
    result = await invalidate_course_cache(r, course_id)
    return result

Production Considerations

1. Environment Configuration

# .env
MONGO_URI=mongodb://localhost:27017/elearning
REDIS_URL=redis://localhost:6379/0
JWT_SECRET=your-super-secret-jwt-key-minimum-32-chars
ACCESS_TOKEN_EXPIRE_MINUTES=15
REFRESH_TOKEN_EXPIRE_DAYS=7
ENVIRONMENT=production
DEBUG=false

2. Redis Configuration

# redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru
save 900 1
save 300 10
save 60 10000

3. Application Startup

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    app.state.mongo_client = create_mongo_client(settings.MONGO_URI)
    app.state.db = app.state.mongo_client.elearning
    app.state.redis = create_redis_client(settings.REDIS_URL)
    
    # Warm up cache
    await warm_courses_cache(app.state.db, app.state.redis)
    
    yield
    
    # Shutdown
    app.state.mongo_client.close()
    await app.state.redis.close()

app = FastAPI(lifespan=lifespan)

4. Cache Warming Strategy

# services/cache_warming.py
async def warm_courses_cache(db: Database, r: Redis) -> None:
    """Pre-warm cache with popular content"""
    WARM_SORTS = ["recent", "popular", "top_rated"]
    WARM_PAGE_SIZE = 12
    
    base_filters = {}
    q = None
    page = 1
    warmed_ids = set()

    # Warm cache for each sort option
    for sort_by in WARM_SORTS:
        page_payload = await list_courses(
            db, r, q=q, filters=base_filters, 
            page=page, page_size=WARM_PAGE_SIZE, sort_by=sort_by
        )
        
        # Collect course IDs from results
        for item in page_payload.get("items", []):
            cid = item.get("_id")
            if cid:
                warmed_ids.add(cid)
    
    # Warm individual course caches
    for cid in warmed_ids:
        await get_course(db, r, cid)

5. Dependencies

# requirements.txt
fastapi
uvicorn[standard]
python-jose[cryptography]
passlib[argon2]
python-dotenv
pydantic[email]
pydantic-settings==2.3.4
pymongo==4.8.0
redis==5.0.4
PyJWT==2.8.0
passlib[bcrypt]==1.7.4
python-multipart

Performance Benchmarks

Real-World Speed Tests

Based on production measurements from the E-Learning platform:

Course Retrieval Performance:
┌─────────────────┬──────────────┬─────────────────┐
│ Cache Layer     │ Response Time│ Requests/Second │
├─────────────────┼──────────────┼─────────────────┤
│ L1 (Memory)     │ 0.5ms        │ 120,000         │
│ L2 (Redis)      │ 2.1ms        │ 45,000          │
│ L3 (Database)   │ 47ms         │ 850             │
└─────────────────┴──────────────┴─────────────────┘

Course Listing Performance:
┌─────────────────┬──────────────┬─────────────────┐
│ Cache Layer     │ Response Time│ Requests/Second │
├─────────────────┼──────────────┼─────────────────┤
│ L1 (Memory)     │ 1.2ms        │ 85,000          │
│ L2 (Redis)      │ 4.8ms        │ 25,000          │
│ L3 (Database)   │ 180ms        │ 200             │
└─────────────────┴──────────────┴─────────────────┘

Authentication Performance:
┌─────────────────┬──────────────┬─────────────────┐
│ Operation       │ Response Time│ Improvement     │
├─────────────────┼──────────────┼─────────────────┤
│ Token Validation│ 0.3ms        │ 150x faster     │
│ Session Lookup  │ 0.8ms        │ 80x faster      │
│ User Data Fetch │ 1.1ms        │ 95x faster      │
└─────────────────┴──────────────┴─────────────────┘

Cache Hit Ratio Impact

• 90%+ Hit Ratio: Average response time 2ms
• 70-89% Hit Ratio: Average response time 8ms
• <70% Hit Ratio: Average response time 25ms

Key Takeaways

1. Layer Strategy: Use memory for hot data, Redis for distributed caching
2. TTL Management: Shorter TTLs for frequently changing data
3. Cache Stampede Protection: Use locks to prevent concurrent database hits
4. Smart Invalidation: Invalidate related data when updates occur
5. Monitoring: Track hit ratios and performance metrics
6. Graceful Degradation: Always have fallback to database

Expected Performance Gains

• Database Load Reduction: 85-95%
• Response Time Improvement: 50-150x faster
• Throughput Increase: 100-500x higher
• Server Cost Savings: 60-80% reduction

This multi-level caching implementation provides excellent performance while maintaining data consistency. The combination of in-memory and Redis caching offers the best of both worlds: ultra-fast access for hot data and distributed caching for scalability.

Remember to monitor your cache hit ratios and adjust TTLs based on your application’s usage patterns. A well-implemented caching strategy can reduce database load by 80-90% while significantly improving response times.