Cache-aside (Lazy Loading)

Overview

Cache-aside is the most common caching pattern. It's called "lazy loading" because you only cache what the application asks for — you don't preload everything. The logic is direct: when a request arrives, you check Redis first. If it's there, you return it (a cache hit). If not, you query PostgreSQL, store it in Redis, and return it (a cache miss). The next request for the same data will be a cache hit.

It's the default pattern REST APIs use in production to cache read endpoints. If you have 100 endpoints in your API and only get budget to optimize caching on 10, cache-aside applied to the most-read ones gives you 80% of the benefit. It covers product catalogs, user profiles, search results, public configuration — everything that's read a lot and changes little.

But cache-aside has a cost known as the cold cache problem: the first request is always slow because it has to query the DB. If your app just started up, or if you use aggressive TTLs, you're going to have a lot of cache misses at first. You'll learn to measure it, to understand when it matters, and to solve it with cache warming. By the end of the capsule you'll have cached a real FastAPI endpoint with a TTL, measured the hit rate before and after, and seen the latency difference in numbers, not in intuition.

The pattern, step by step

┌────────────┐
│   Client   │
└─────┬──────┘
      │ GET /products
      ▼
┌────────────┐         ┌────────────┐
│ FastAPI    │────────▶│   Redis    │
│ Endpoint   │  GET    │            │
└─────┬──────┘ cache:  └────┬───────┘
      │        products      │
      │                      │
      │     ┌────────────────┘
      │     ▼
      │   Hit?
      │     │
      │     ├── Yes → return cached response (1ms)
      │     │
      │     └── No
      │          │
      ▼          ▼
┌────────────┐         ┌────────────┐
│ FastAPI    │────────▶│ PostgreSQL │
│ Endpoint   │  SELECT │            │
└─────┬──────┘         └────────────┘
      │
      │ ◄────── result (50ms)
      │
      ▼
┌────────────┐
│   Redis    │
│ SET cache: │
│ products   │
│ EX 300     │
└────────────┘
      │
      ▼
   return response (51ms)

Pseudocode

def get_data(key):
    cached = redis.get(key)

    if cached is not None:
        return cached  # cache hit, ~1ms

    # cache miss
    data = database.query(key)  # ~50ms
    redis.set(key, data, ex=TTL_SECONDS)
    return data

It's that simple. All of cache-aside's complexity lives in the decisions surrounding this code: what to cache, with what TTL, how to invalidate.

Implementation with redis-py

We're going to build cache-aside step by step. We start with the simplest version and improve it up to production quality.

Workspace setup

cd ~/projects/redis-guide/module-02-patterns
source .venv/bin/activate
mkdir cache-aside-demo && cd cache-aside-demo

Create requirements.txt:

redis>=7.4
fastapi>=0.136
uvicorn[standard]>=0.27.0
sqlalchemy>=2.0

pip install -r requirements.txt

Version 1: The basic implementation

Create cache_aside_v1.py:

"""
Cache-aside V1: the simplest possible implementation.
It demonstrates the flow: check cache → miss → query DB → store → return.
"""
import time
import json
import redis


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


# We simulate PostgreSQL with an in-memory dict
# In production, this would be SQLAlchemy with real queries
fake_db = {
    1: {"id": 1, "name": "MacBook Pro", "price": 2499, "stock": 10},
    2: {"id": 2, "name": "ThinkPad X1", "price": 1899, "stock": 5},
    3: {"id": 3, "name": "Dell XPS 13", "price": 1599, "stock": 8},
}


def query_db(product_id: int) -> dict | None:
    """
    Simulates a query to PostgreSQL.
    We add an artificial delay to simulate network latency + query time.
    """
    time.sleep(0.05)  # 50ms - typical for a query with a JOIN in PostgreSQL
    return fake_db.get(product_id)


def get_product(product_id: int) -> dict | None:
    """Cache-aside: check cache → miss → query DB → store → return."""
    cache_key = f"product:{product_id}"

    # 1. Check the cache
    cached = r.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 2. Cache miss → query the DB
    product = query_db(product_id)
    if product is None:
        return None

    # 3. Store it in the cache (no TTL yet, we'll see that in V2)
    r.set(cache_key, json.dumps(product))

    return product


if __name__ == "__main__":
    # First request: cache miss
    start = time.time()
    p1 = get_product(1)
    elapsed = (time.time() - start) * 1000
    print(f"Request 1 (miss): {elapsed:.1f}ms - {p1}")

    # Second request: cache hit
    start = time.time()
    p2 = get_product(1)
    elapsed = (time.time() - start) * 1000
    print(f"Request 2 (hit):  {elapsed:.1f}ms - {p2}")

    # Third request: cache hit
    start = time.time()
    p3 = get_product(1)
    elapsed = (time.time() - start) * 1000
    print(f"Request 3 (hit):  {elapsed:.1f}ms - {p3}")

Run it:

python cache_aside_v1.py

Expected output:

Request 1 (miss): 51.3ms - {'id': 1, 'name': 'MacBook Pro', 'price': 2499, 'stock': 10}
Request 2 (hit):  1.2ms - {'id': 1, 'name': 'MacBook Pro', 'price': 2499, 'stock': 10}
Request 3 (hit):  0.9ms - {'id': 1, 'name': 'MacBook Pro', 'price': 2499, 'stock': 10}

What happened:

• Request 1: the cache is empty → query the "DB" (50 ms) → store it in Redis → return (a total of ~51 ms)
• Requests 2-3: the cache has the key → read from Redis (1 ms) → return

Speedup: 51 ms → 1 ms = ~50x faster.

But there's a problem: this version never expires the cache. If you modify a product in the DB, you'll serve stale data forever.

Version 2: With a TTL

Add a TTL so Redis cleans up automatically:

"""
Cache-aside V2: with a TTL.
"""
import time
import json
import redis


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


CACHE_TTL_SECONDS = 60  # the cache lives for 1 minute

fake_db = {
    1: {"id": 1, "name": "MacBook Pro", "price": 2499, "stock": 10},
    2: {"id": 2, "name": "ThinkPad X1", "price": 1899, "stock": 5},
}


def query_db(product_id: int) -> dict | None:
    time.sleep(0.05)
    return fake_db.get(product_id)


def get_product(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"

    cached = r.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    product = query_db(product_id)
    if product is None:
        return None

    # ✨ The key change: SET with a TTL
    r.set(cache_key, json.dumps(product), ex=CACHE_TTL_SECONDS)

    return product


if __name__ == "__main__":
    # Fill the cache
    print("First request (miss):")
    p1 = get_product(1)
    print(f"  Result: {p1['name']}")
    print(f"  TTL: {r.ttl('product:1')} seconds remaining")

    # Verify a cache hit
    print("\nSecond request (hit):")
    p2 = get_product(1)
    print(f"  Result: {p2['name']}")
    print(f"  TTL: {r.ttl('product:1')} seconds remaining")

    # Simulate the passage of time: delete the cache by hand
    print("\nWe simulate that 60 sec went by (the cache expires)...")
    r.delete('product:1')

    # A cache miss again
    print("\nThird request (a miss again):")
    p3 = get_product(1)
    print(f"  Result: {p3['name']}")

Output:

First request (miss):
  Result: MacBook Pro
  TTL: 60 seconds remaining

Second request (hit):
  Result: MacBook Pro
  TTL: 58 seconds remaining

We simulate that 60 sec went by (the cache expires)...

Third request (a miss again):
  Result: MacBook Pro

This version is functional for simple production use. But one critical thing is missing: what happens if Redis goes down?

Version 3: With error handling (graceful degradation)

"""
Cache-aside V3: with error handling.
The API has to keep working if Redis goes down (slower, but alive).
"""
import time
import json
import logging
import redis
from redis.exceptions import RedisError, ConnectionError as RedisConnectionError


logger = logging.getLogger(__name__)


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


CACHE_TTL_SECONDS = 60

fake_db = {
    1: {"id": 1, "name": "MacBook Pro", "price": 2499, "stock": 10},
    2: {"id": 2, "name": "ThinkPad X1", "price": 1899, "stock": 5},
}


def query_db(product_id: int) -> dict | None:
    time.sleep(0.05)
    return fake_db.get(product_id)


def get_product(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"

    # 1. Try the cache (if Redis fails, we ignore it and go to the DB)
    try:
        cached = r.get(cache_key)
        if cached is not None:
            return json.loads(cached)
    except (RedisError, RedisConnectionError) as e:
        logger.warning(f"Redis read failed for {cache_key}: {e}")
        # We don't return an error — we fall through to the DB

    # 2. Query the DB (always available — the source of truth)
    product = query_db(product_id)
    if product is None:
        return None

    # 3. Try caching it (if Redis fails, we ignore it)
    try:
        r.set(cache_key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    except (RedisError, RedisConnectionError) as e:
        logger.warning(f"Redis write failed for {cache_key}: {e}")
        # We don't return an error — the data is returned anyway, it just wasn't cached

    return product


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    p = get_product(1)
    print(f"Product: {p}")

    # Simulate Redis being down: stop Redis with docker stop redis-dev
    # and run the script again.
    # The application should keep working, showing warnings.

To test it:

# In another terminal, stop Redis:
docker stop redis-dev

# Run the script:
python cache_aside_v3.py

# Output:
# WARNING:__main__:Redis read failed for product:1: ...
# WARNING:__main__:Redis write failed for product:1: ...
# Product: {'id': 1, 'name': 'MacBook Pro', ...}

# Restart Redis:
docker start redis-dev

The critical part: the application returns the product correctly even with Redis down. It just logs warnings and degrades to "no cache mode" temporarily.

This is the version that goes to production. V1 and V2 are educational; V3 is the one that scales.

The cold cache problem

The "cold cache problem" is cache-aside's main downside: the first request for any piece of data is always slow because the cache is empty.

When it's a real problem

A minor problem (don't worry about it):

• Your app has continuous traffic. After the first hit, all the following ones are fast
• The requests are spread out over time. A cold cache every 5 minutes doesn't register with the user

A serious problem:

• After a deploy: every worker restarts with an empty cache. If you have 1000 popular products, the first 1000 requests to the endpoint are misses → PostgreSQL saturates
• After a Redis outage: the entire cache is lost, and you're back to a global cold cache
• A traffic spike when the TTL expires: if all the caches expire at the same time, every subsequent request is a simultaneous miss (this is cache stampede, covered in capsule 05)

Demonstrating the cold cache problem

Create cold_cache_demo.py:

"""
Demonstrates the cold cache problem.
"""
import time
import json
import redis
import statistics


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


fake_db = {i: {"id": i, "name": f"Product {i}"} for i in range(1, 101)}


def query_db(product_id: int) -> dict | None:
    time.sleep(0.05)  # 50ms
    return fake_db.get(product_id)


def get_product(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    product = query_db(product_id)
    if product is None:
        return None

    r.set(cache_key, json.dumps(product), ex=300)
    return product


def benchmark(label: str, num_requests: int):
    """Measures the latency of N requests."""
    latencies = []
    for i in range(num_requests):
        product_id = (i % 10) + 1  # we rotate among 10 products
        start = time.time()
        get_product(product_id)
        latencies.append((time.time() - start) * 1000)

    print(f"\n{label} ({num_requests} requests):")
    print(f"  Min: {min(latencies):.1f}ms")
    print(f"  Max: {max(latencies):.1f}ms")
    print(f"  Avg: {statistics.mean(latencies):.1f}ms")
    print(f"  p50: {statistics.median(latencies):.1f}ms")
    print(f"  p95: {sorted(latencies)[int(len(latencies)*0.95)]:.1f}ms")


if __name__ == "__main__":
    # Clear the cache so we start fresh
    for i in range(1, 11):
        r.delete(f"product:{i}")

    # Cold cache: the first N requests are slow
    benchmark("COLD CACHE (empty cache)", 100)

    # Warm cache: the same requests, but now with a full cache
    benchmark("WARM CACHE (populated cache)", 100)

Run it:

python cold_cache_demo.py

Expected output:

COLD CACHE (empty cache) (100 requests):
  Min: 0.8ms
  Max: 53.1ms
  Avg: 6.4ms
  p50: 1.2ms
  p95: 51.5ms

WARM CACHE (populated cache) (100 requests):
  Min: 0.4ms
  Max: 2.1ms
  Avg: 0.8ms
  p50: 0.7ms
  p95: 1.3ms

What it demonstrates:

• With a cold cache, the first 10 requests (one per product) are ~50 ms each (cache miss + DB query)
• The next 90 are fast hits
• With a warm cache, they're all hits — an average of ~1 ms

The p95 (95th percentile) reveals the reality: with a cold cache, 5% of your users will see >50 ms of latency. With a warm cache, everyone sees <1.5 ms.

Cache warming: the solution to the cold cache problem

Cache warming is preloading popular data when the application starts, before it receives real traffic.

The strategy: preload at startup

"""
Cache warming: loading popular products when the app starts.
"""
import time
import json
import redis


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


# We simulate the top 20 most-viewed products (in production, a query to an analytics DB)
top_products_ids = list(range(1, 21))


fake_db = {i: {"id": i, "name": f"Product {i}", "price": 100 + i*10} for i in range(1, 101)}


def query_db_batch(product_ids: list[int]) -> dict[int, dict]:
    """A batch query to PostgreSQL: 1 query with WHERE id IN (...)."""
    time.sleep(0.1)  # A batch query for 20 IDs is ~100ms
    return {pid: fake_db[pid] for pid in product_ids if pid in fake_db}


def warm_cache():
    """Preload the popular products at startup."""
    print(f"Warming the cache with {len(top_products_ids)} popular products...")
    start = time.time()

    products = query_db_batch(top_products_ids)

    # A pipeline for the batch SET
    pipe = r.pipeline()
    for pid, product in products.items():
        pipe.set(f"product:{pid}", json.dumps(product), ex=300)
    pipe.execute()

    elapsed = (time.time() - start) * 1000
    print(f"Cache warmed with {len(products)} products in {elapsed:.1f}ms")


if __name__ == "__main__":
    # Clear the cache
    for i in top_products_ids:
        r.delete(f"product:{i}")

    # Warm it
    warm_cache()

    # Now every request for a top product is a hit
    print("\nVerifying that the products are cached:")
    for pid in top_products_ids[:5]:
        cached = r.get(f"product:{pid}")
        print(f"  product:{pid} → {'cached ✓' if cached else 'NOT cached ✗'}")

Output:

Warming the cache with 20 popular products...
Cache warmed with 20 products in 102.3ms

Verifying that the products are cached:
  product:1 → cached ✓
  product:2 → cached ✓
  product:3 → cached ✓
  ...

Integrating with the FastAPI lifespan:

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: warm the cache
    warm_cache()
    print("App ready, cache warmed")

    yield

    # Shutdown: clean up if necessary
    print("App shutting down")


app = FastAPI(lifespan=lifespan)

When the app starts, before it accepts requests, the cache already has the popular products loaded. Cold cache problem mitigated.

Hit rate and latency: real measurement

You're going to implement hit rate tracking by hand. Later (in module 5) you'll integrate this with Prometheus/Grafana, but the basic tracking is this:

"""
Tracking hit rate and latency with Redis counters.
"""
import time
import json
import redis


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

fake_db = {i: {"id": i, "name": f"Product {i}"} for i in range(1, 51)}


def query_db(product_id: int) -> dict | None:
    time.sleep(0.05)
    return fake_db.get(product_id)


def get_product_with_metrics(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"

    cached = r.get(cache_key)
    if cached is not None:
        # HIT: increment the counter
        r.incr("metrics:cache:hits")
        return json.loads(cached)

    # MISS: increment the counter
    r.incr("metrics:cache:misses")

    product = query_db(product_id)
    if product is None:
        return None

    r.set(cache_key, json.dumps(product), ex=300)
    return product


def report_metrics():
    hits = int(r.get("metrics:cache:hits") or 0)
    misses = int(r.get("metrics:cache:misses") or 0)
    total = hits + misses

    if total == 0:
        print("No requests yet")
        return

    hit_rate = hits / total * 100

    print(f"\n=== Cache metrics ===")
    print(f"  Total requests: {total}")
    print(f"  Hits:   {hits} ({hit_rate:.1f}%)")
    print(f"  Misses: {misses} ({100-hit_rate:.1f}%)")


if __name__ == "__main__":
    # Clean up
    r.delete("metrics:cache:hits", "metrics:cache:misses")
    for i in range(1, 11):
        r.delete(f"product:{i}")

    # Simulate realistic traffic: 80% of the requests go to the top 5 products
    import random
    for _ in range(100):
        if random.random() < 0.8:
            product_id = random.randint(1, 5)  # the top 5 (highly repeated)
        else:
            product_id = random.randint(1, 50)  # any other one

        get_product_with_metrics(product_id)

    report_metrics()

Typical output:

=== Cache metrics ===
  Total requests: 100
  Hits:   85 (85.0%)
  Misses: 15 (15.0%)

An 85% hit rate is very good. It means 85% of the requests never touch PostgreSQL.

Interpreting the hit rate

• <50%: A poorly designed cache, a TTL that's too short, or keys that aren't reused enough
• 50-80%: Acceptable but there's room
• 80-95%: Well designed for a typical API
• >95%: Excellent — but check that the TTL isn't excessive (the data could be very stale)

A real case: caching a FastAPI endpoint

Create app.py — a complete FastAPI endpoint with cache-aside:

"""
A FastAPI app with cache-aside on a products endpoint.
"""
import time
import json
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
import redis
from redis.exceptions import RedisError


logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


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


CACHE_TTL = 300  # 5 minutes


# We simulate PostgreSQL
fake_db = {
    i: {
        "id": i,
        "name": f"Product {i}",
        "price": 100 + i * 10,
        "stock": (i * 7) % 50,
        "category": "electronics" if i % 2 == 0 else "books",
    }
    for i in range(1, 101)
}


def query_db_get_product(product_id: int) -> dict | None:
    """Simulates a SELECT query to PostgreSQL."""
    time.sleep(0.05)  # 50ms of simulated latency
    return fake_db.get(product_id)


def query_db_list_products() -> list[dict]:
    """Simulates SELECT * FROM products."""
    time.sleep(0.1)  # 100ms — a heavier query
    return list(fake_db.values())


def warm_cache():
    """Preloads the top 10 products at startup."""
    logger.info("Warming cache...")
    pipe = r.pipeline()
    for pid in range(1, 11):
        product = fake_db[pid]
        pipe.set(f"product:{pid}", json.dumps(product), ex=CACHE_TTL)
    pipe.execute()
    logger.info(f"Cache warmed with {10} products")


@asynccontextmanager
async def lifespan(app: FastAPI):
    warm_cache()
    yield


app = FastAPI(title="Cached Products API", lifespan=lifespan)


@app.get("/products/{product_id}")
def get_product(product_id: int):
    """An endpoint with cache-aside + graceful degradation."""
    cache_key = f"product:{product_id}"

    # Try the cache
    try:
        cached = r.get(cache_key)
        if cached is not None:
            r.incr("metrics:hits")
            return json.loads(cached)
    except RedisError as e:
        logger.warning(f"Redis read failed: {e}")

    # Cache miss → query the DB
    product = query_db_get_product(product_id)
    if product is None:
        raise HTTPException(404, "Product not found")

    r.incr("metrics:misses")

    # Try caching it
    try:
        r.set(cache_key, json.dumps(product), ex=CACHE_TTL)
    except RedisError as e:
        logger.warning(f"Redis write failed: {e}")

    return product


@app.get("/metrics/cache")
def cache_metrics():
    """An endpoint for inspecting the metrics."""
    try:
        hits = int(r.get("metrics:hits") or 0)
        misses = int(r.get("metrics:misses") or 0)
        total = hits + misses
        hit_rate = (hits / total * 100) if total > 0 else 0

        return {
            "total_requests": total,
            "hits": hits,
            "misses": misses,
            "hit_rate_percent": round(hit_rate, 2),
        }
    except RedisError:
        return {"error": "Redis unavailable"}

Run it:

uvicorn app:app --reload

Try it:

# First request: cache miss (~50ms)
time curl -s http://localhost:8000/products/1 | python -m json.tool

# Second request: cache hit (~1ms)
time curl -s http://localhost:8000/products/1 | python -m json.tool

# View the metrics
curl -s http://localhost:8000/metrics/cache | python -m json.tool

Expected output from /metrics/cache:

{
  "total_requests": 2,
  "hits": 1,
  "misses": 1,
  "hit_rate_percent": 50.0
}

If you make 100 requests for product 1, you should see a hit rate of ~99%. The first one was a miss; the next 99 were hits.

When NOT to use cache-aside

Cache-aside covers 80% of cases, but not all of them. Here's when to choose another pattern:

Use write-through when:

• The data changed and it CAN'T be shown stale. Example: a bank account balance, role authorization
• Writes are rare compared to reads. If you write once per hour but read 1000 times, write-through avoids misses without a real penalty on writes

Use write-behind when:

• Massive writes that can be batched. Example: analytics events, audit logs
• It's acceptable to lose some data if Redis goes down. For legal auditing, NO; for like counters, fine

Don't use caching at all when:

• The data changes on every request. Example: the current timestamp, a random number
• The data is only read once. If the hit rate would be <10%, the cache's overhead is worse than going straight to the DB
• The data is millisecond-sensitive. If your UX requires immediate consistency with no TTL, adding invalidation complicates more than it saves

Troubleshooting

Problem 1: The hit rate is low (< 50%) but the endpoint is called a lot

Cause: Every request uses a different key. For example, caching GET /search?q=... with unique queries.

Solution: Only cache common queries. For search, consider not caching (the space of unique queries is infinite) or cache only the top N queries:

COMMON_QUERIES = {"laptop", "phone", "shoes", "book"}

def search(query: str):
    if query in COMMON_QUERIES:
        # Cache-aside
        cached = r.get(f"search:{query}")
        ...
    else:
        # Skip the cache, query the DB directly
        return db_search(query)

Problem 2: Stale data in the cache after an update

Cause: The POST/PUT/DELETE endpoint doesn't invalidate the cache.

Solution: Invalidate it explicitly:

@app.post("/products")
def create_product(data: ProductCreate):
    product = db.create(data)
    r.delete("product:list")  # invalidate the list
    return product

@app.put("/products/{id}")
def update_product(id: int, data: ProductUpdate):
    product = db.update(id, data)
    r.delete(f"product:{id}")  # invalidate the specific one
    r.delete("product:list")    # invalidate the list (it changed too)
    return product

Capsule 04 covers invalidation in detail.

Problem 3: A very short TTL = a low hit rate

Cause: A 30-second TTL means you regenerate the cache every minute. If your data changes every hour, the optimal TTL is ~30-50 minutes.

Solution: Match the TTL to how fast the data changes:

Problem 4: Serialization errors (JSON doesn't support datetime)

Cause: datetime and other Python objects aren't JSON serializable.

Solution: Use a custom encoder:

import json
from datetime import datetime

def json_serial(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Type {type(obj)} not serializable")

# When storing:
r.set(key, json.dumps(data, default=json_serial))

# When reading, you parse the ISO string:
data = json.loads(cached)
data["created_at"] = datetime.fromisoformat(data["created_at"])

An alternative: use pickle if your data is more complex, but then set decode_responses=False and handle bytes.

Problem 5: A cache miss spike after a deploy

Cause: Every worker restarts with an empty cache. A global cold cache.

Solution: Implement cache warming in a lifespan event (you saw it above). For large apps, consider loading the cache from a previous Redis dump (see BGSAVE and RDB).

Problem 6: Memory usage grows without limit

Cause: You cache without a memory limit and the TTLs aren't aggressive enough.

Solution:

1. Configure maxmemory in redis.conf (or with a Docker flag): redis-server --maxmemory 1gb --maxmemory-policy allkeys-lru
2. The allkeys-lru policy evicts the least-used keys when it fills up
3. Lower the TTLs if you can

docker run -d --name redis-dev -p 6379:6379 \
  redis:7 redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

Exercises

Exercise 1: Implement basic cache-aside (Easy)

Implement get_user(user_id) with cache-aside. Use a user:{user_id} key with a 5-minute TTL. Simulate the "DB" with a dict. Measure the time of the first and second call.

import time
import json
import redis

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

fake_users_db = {
    1: {"id": 1, "name": "Alice", "email": "alice@x.com"},
    2: {"id": 2, "name": "Bob", "email": "bob@x.com"},
}


def query_db_user(user_id):
    time.sleep(0.05)  # 50ms simulated query
    return fake_users_db.get(user_id)


def get_user(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    user = query_db_user(user_id)
    if user is None:
        return None

    r.set(cache_key, json.dumps(user), ex=300)
    return user


# Test
start = time.time()
u1 = get_user(1)
print(f"First call: {(time.time() - start) * 1000:.1f}ms - {u1}")

start = time.time()
u2 = get_user(1)
print(f"Second call: {(time.time() - start) * 1000:.1f}ms - {u2}")

First call: 51.2ms - {'id': 1, 'name': 'Alice', 'email': 'alice@x.com'}
Second call: 0.9ms - {'id': 1, 'name': 'Alice', 'email': 'alice@x.com'}

Exercise 2: Hit rate tracking (Easy-Medium)

Modify exercise 1 to increment metrics:user:hits and metrics:user:misses. After 100 calls varying user_id between 1 and 5, calculate the hit rate.

import time, json, random, redis

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

fake_users_db = {i: {"id": i, "name": f"User {i}"} for i in range(1, 6)}


def query_db_user(user_id):
    time.sleep(0.01)
    return fake_users_db.get(user_id)


def get_user_metrics(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        r.incr("metrics:user:hits")
        return json.loads(cached)

    r.incr("metrics:user:misses")
    user = query_db_user(user_id)
    if user:
        r.set(cache_key, json.dumps(user), ex=300)
    return user


# Reset counters
r.delete("metrics:user:hits", "metrics:user:misses")
for i in range(1, 6):
    r.delete(f"user:{i}")

# 100 calls with a random user_id between 1-5
for _ in range(100):
    get_user_metrics(random.randint(1, 5))

hits = int(r.get("metrics:user:hits") or 0)
misses = int(r.get("metrics:user:misses") or 0)
total = hits + misses
print(f"Hits: {hits}, Misses: {misses}, Hit rate: {hits/total*100:.1f}%")

Hits: 95, Misses: 5, Hit rate: 95.0%

Exercise 3: Graceful degradation (Medium)

Modify get_user so it works with Redis down. Stop Redis with docker stop redis-dev, run your script, and verify that it keeps returning data correctly. Restart Redis.

import time, json, logging, redis
from redis.exceptions import RedisError

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

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

fake_users_db = {1: {"id": 1, "name": "Alice"}}


def query_db_user(user_id):
    time.sleep(0.05)
    return fake_users_db.get(user_id)


def get_user_safe(user_id):
    cache_key = f"user:{user_id}"

    # Try the cache
    try:
        cached = r.get(cache_key)
        if cached:
            return json.loads(cached)
    except RedisError as e:
        logger.warning(f"Redis read failed: {e}")

    # Always query the DB if there's no cache hit (or Redis is down)
    user = query_db_user(user_id)
    if user is None:
        return None

    # Try caching it
    try:
        r.set(cache_key, json.dumps(user), ex=300)
    except RedisError as e:
        logger.warning(f"Redis write failed: {e}")

    return user


# Test:
# 1. With Redis running: it works normally
# 2. docker stop redis-dev → run the script → you see the warnings but the user is returned correctly
# 3. docker start redis-dev → back to normal mode

print(get_user_safe(1))

# With Redis up:
python script.py
# Output: {'id': 1, 'name': 'Alice'}

# Stop Redis:
docker stop redis-dev
python script.py
# Output:
# WARNING:__main__:Redis read failed: Connection refused...
# WARNING:__main__:Redis write failed: Connection refused...
# {'id': 1, 'name': 'Alice'}   ← it still works!

docker start redis-dev

Exercise 4: Cache warming (Medium)

Implement a warm_user_cache() function that preloads the 10 most active users when the app starts. Measure the total time and verify that the subsequent queries are hits.

import time, json, redis

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

# We simulate a list of "top 10 users" (in production, a query to a stats table)
TOP_USERS_IDS = [1, 5, 12, 23, 47, 88, 102, 145, 200, 333]


def query_db_users_batch(user_ids):
    """SELECT * FROM users WHERE id IN (...)"""
    time.sleep(0.08)  # A batch query for 10 IDs
    return {uid: {"id": uid, "name": f"User {uid}"} for uid in user_ids}


def warm_user_cache():
    print(f"Warming the cache with the top {len(TOP_USERS_IDS)} users...")
    start = time.time()

    users = query_db_users_batch(TOP_USERS_IDS)

    pipe = r.pipeline()
    for uid, user in users.items():
        pipe.set(f"user:{uid}", json.dumps(user), ex=300)
    pipe.execute()

    elapsed = (time.time() - start) * 1000
    print(f"Cache warmed with {len(users)} users in {elapsed:.1f}ms")


def get_user(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return ("hit", json.loads(cached))

    time.sleep(0.05)  # simulated DB query
    user = {"id": user_id, "name": f"User {user_id}"}
    r.set(cache_key, json.dumps(user), ex=300)
    return ("miss", user)


# Clean up
for uid in TOP_USERS_IDS:
    r.delete(f"user:{uid}")

# Warm it
warm_user_cache()

# Verify: every top user is a hit
print("\nVerifying hits for top users:")
for uid in TOP_USERS_IDS[:5]:
    status, user = get_user(uid)
    print(f"  user:{uid} → {status}")

Warming the cache with the top 10 users...
Cache warmed with 10 users in 82.5ms

Verifying hits for top users:
  user:1 → hit
  user:5 → hit
  user:12 → hit
  user:23 → hit
  user:47 → hit

Exercise 5: Invalidation on update (Medium)

Implement update_user(user_id, data) that updates the "DB" (the dict) AND invalidates the cache. Show that after the update, the next read is a miss + returns the updated data.

import time, json, redis

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

fake_users_db = {1: {"id": 1, "name": "Alice", "age": 30}}


def query_db_user(user_id):
    time.sleep(0.05)
    return fake_users_db.get(user_id)


def get_user(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return ("hit", json.loads(cached))

    user = query_db_user(user_id)
    r.set(cache_key, json.dumps(user), ex=300)
    return ("miss", user)


def update_user(user_id, data):
    """Updates the DB and invalidates the cache."""
    if user_id not in fake_users_db:
        return None

    fake_users_db[user_id].update(data)

    # ✨ Invalidate the cache
    r.delete(f"user:{user_id}")

    return fake_users_db[user_id]


# Clean up
r.delete("user:1")

# Read 1: miss (it loads into the cache)
status, u = get_user(1)
print(f"Read 1: {status} - {u}")

# Read 2: hit
status, u = get_user(1)
print(f"Read 2: {status} - {u}")

# Update
print("\nUpdate user:1 → age=31")
update_user(1, {"age": 31})

# Read 3: miss (because we invalidated it), returns the updated data
status, u = get_user(1)
print(f"Read 3: {status} - {u}")

# Read 4: hit
status, u = get_user(1)
print(f"Read 4: {status} - {u}")

Read 1: miss - {'id': 1, 'name': 'Alice', 'age': 30}
Read 2: hit - {'id': 1, 'name': 'Alice', 'age': 30}

Update user:1 → age=31

Read 3: miss - {'id': 1, 'name': 'Alice', 'age': 31}
Read 4: hit - {'id': 1, 'name': 'Alice', 'age': 31}

Exercise 6: Benchmarking cold vs warm cache (Hard)

Write a script that runs 1000 requests against the get_product() endpoint with random products between IDs 1-50. Measure the p50, p95, and p99 latency in two scenarios:

1. Cold cache (after FLUSHDB)
2. Warm cache (after warming with the top 50)

Compare the results.

import time, json, random, redis, statistics

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

fake_db = {i: {"id": i, "name": f"Product {i}"} for i in range(1, 51)}


def query_db(pid):
    time.sleep(0.05)
    return fake_db.get(pid)


def get_product(pid):
    cache_key = f"product:{pid}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    p = query_db(pid)
    if p:
        r.set(cache_key, json.dumps(p), ex=300)
    return p


def warm_cache(ids):
    pipe = r.pipeline()
    for pid in ids:
        if pid in fake_db:
            pipe.set(f"product:{pid}", json.dumps(fake_db[pid]), ex=300)
    pipe.execute()


def benchmark(label, num_requests=1000):
    latencies = []
    for _ in range(num_requests):
        pid = random.randint(1, 50)
        start = time.time()
        get_product(pid)
        latencies.append((time.time() - start) * 1000)

    sorted_lat = sorted(latencies)
    print(f"\n{label}:")
    print(f"  p50: {sorted_lat[int(len(sorted_lat)*0.50)]:.1f}ms")
    print(f"  p95: {sorted_lat[int(len(sorted_lat)*0.95)]:.1f}ms")
    print(f"  p99: {sorted_lat[int(len(sorted_lat)*0.99)]:.1f}ms")
    print(f"  avg: {statistics.mean(latencies):.1f}ms")
    print(f"  max: {max(latencies):.1f}ms")


# Cold cache
r.flushdb()
benchmark("COLD CACHE", 1000)

# Warm + benchmark
r.flushdb()
warm_cache(range(1, 51))
benchmark("WARM CACHE", 1000)

COLD CACHE:
  p50: 1.2ms
  p95: 51.5ms
  p99: 53.0ms
  avg: 4.5ms
  max: 53.4ms

WARM CACHE:
  p50: 0.7ms
  p95: 1.4ms
  p99: 2.0ms
  avg: 0.9ms
  max: 3.2ms

Summary

In this capsule you learned:

• Cache-aside (lazy loading) is the most common caching pattern: check cache → miss → query DB → store → return
• 3 implementation versions:
  • V1: the basic one with GET/SET
  • V2: with a TTL (SET ... EX seconds)
  • V3: with graceful degradation (try/except around Redis errors)
• The cold cache problem: the first request for each piece of data is slow. It matters after deploys, Redis restarts, or mass expirations
• Cache warming: preloading popular data at startup with the FastAPI lifespan + pipelines
• Metric tracking: hit rate (>80% is good) and latency (p95, p99) reveal your cache's reality
• When NOT to use cache-aside: data that changes on every request, single-use data, millisecond-sensitive data
• Implementation with FastAPI: integrating with lifespan for warming, a /metrics/cache endpoint for observability

The key decision: cache-aside is the default. Use it for 80% of your reads. When you need strong consistency (write-through) or batch writes (write-behind), capsule 03 takes over.

Additional resources

1. Microsoft Cloud Design Patterns: Cache-Aside — The pattern formalized, with a diagram
2. AWS: Caching Strategies — Lazy Loading — Cache-aside from the AWS ElastiCache perspective
3. Redis: Cache Strategies — The official implementation with redis-py
4. Performance Calendar: Caching at Scale — Caching in large systems (Twitter, Facebook scale)
5. The Twelve-Factor App: Backing Services — Why Redis should be treated as a backing service
6. redis-py: Connection Errors — How to handle errors correctly

What's next?

In Capsule 03 you leave cache-aside behind and get into the other two patterns: write-through (writing synchronously to the DB and the cache, guaranteeing consistency) and write-behind (writing to the cache, with an async batch to the DB — maximum performance but with risk). You'll implement both, compare them with cache-aside, and learn to decide which one to use for each type of data.

Keep the app you built running. Capsule 03 expands on the same project.