Implement the Cache-Aside Pattern with Redis

Cache-aside (also called lazy loading) is the most common caching pattern. The application checks the cache first; on a miss, it loads from the database, writes the result to the cache, and returns it. On writes, the application updates the database and invalidates the cache entry. Here is a working implementation of cache-aside in Python with Redis, including thundering herd protection and write-through invalidation.

When to Use This

• Read-heavy workloads where the same data is accessed repeatedly
• Data that changes infrequently but is expensive to fetch
• You want the cache to be optional — the system works without it

Prerequisites

• Python 3.10+
• redis package (pip install redis)
• A database client (SQLAlchemy, Psycopg, or similar)

Solution

1. Install Dependencies

pip install redis

2. Implement Cache-Aside Read

import json
import logging
from redis import Redis

logger = logging.getLogger(__name__)


class CacheAside:
    def __init__(self, redis_client: Redis, default_ttl: int = 300):
        self.redis = redis_client
        self.default_ttl = default_ttl

    def get_or_load(
        self,
        key: str,
        loader: callable,
        ttl: int | None = None,
    ) -> dict | None:
        """Read from cache, or load from source and cache the result.

        Args:
            key: Cache key.
            loader: Function that loads data from the source on cache miss.
            ttl: Override default TTL in seconds.

        Returns:
            Cached or freshly loaded data.
        """
        try:
            cached = self.redis.get(key)
            if cached is not None:
                return json.loads(cached)
        except Exception as e:
            logger.warning("Cache read failed for %s: %s", key, e)

        result = loader()
        if result is None:
            return None

        try:
            self.redis.setex(key, ttl or self.default_ttl, json.dumps(result, default=str))
        except Exception as e:
            logger.warning("Cache write failed for %s: %s", key, e)

        return result

    def invalidate(self, key: str) -> None:
        """Remove a key from the cache."""
        try:
            self.redis.delete(key)
        except Exception as e:
            logger.warning("Cache invalidation failed for %s: %s", key, e)

    def invalidate_pattern(self, pattern: str) -> int:
        """Delete all keys matching a pattern (e.g., 'user:*')."""
        count = 0
        try:
            for key in self.redis.scan_iter(match=pattern, count=100):
                self.redis.delete(key)
                count += 1
        except Exception as e:
            logger.warning("Pattern invalidation failed for %s: %s", pattern, e)
        return count

3. Implement Write-Through Invalidation

from myapp.database import db

cache = CacheAside(redis_client, default_ttl=300)

def get_user(user_id: str) -> dict | None:
    return cache.get_or_load(
        key=f"user:{user_id}",
        loader=lambda: db.users.find_by_id(user_id),
        ttl=120,
    )

def update_user(user_id: str, data: dict) -> dict:
    user = db.users.update(user_id, data)
    cache.invalidate(f"user:{user_id}")
    return user

def delete_user(user_id: str) -> None:
    db.users.delete(user_id)
    cache.invalidate(f"user:{user_id}")
    cache.invalidate_pattern(f"user_posts:{user_id}:*")

4. Thundering Herd Protection

When a cache entry expires, multiple requests may simultaneously trigger the loader, causing a database spike. A lock prevents this:

import time
import uuid

class CacheAsideWithLock(CacheAside):
    def get_or_load(
        self,
        key: str,
        loader: callable,
        ttl: int | None = None,
        lock_timeout: int = 10,
    ) -> dict | None:
        # Try cache first
        try:
            cached = self.redis.get(key)
            if cached is not None:
                return json.loads(cached)
        except Exception:
            pass

        # Acquire lock to prevent thundering herd
        lock_key = f"lock:{key}"
        lock_token = str(uuid.uuid4())

        acquired = self.redis.set(lock_key, lock_token, nx=True, ex=lock_timeout)
        if acquired:
            try:
                result = loader()
                if result is not None:
                    self.redis.setex(key, ttl or self.default_ttl, json.dumps(result, default=str))
                return result
            finally:
                # Release lock only if we still own it
                self._release_lock(lock_key, lock_token)
        else:
            # Wait and retry
            for _ in range(5):
                time.sleep(0.1)
                cached = self.redis.get(key)
                if cached is not None:
                    return json.loads(cached)
            # Lock timed out — load anyway as fallback
            return loader()

    def _release_lock(self, lock_key: str, token: str) -> None:
        """Release lock using Lua script to prevent releasing someone else's lock."""
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(script, 1, lock_key, token)

5. Cache Warming

Pre-populate the cache before traffic hits:

def warm_user_cache(user_ids: list[str]) -> None:
    users = db.users.find_many({"id": {"in": user_ids}})
    for user in users:
        cache_key = f"user:{user['id']}"
        redis_client.setex(cache_key, 300, json.dumps(user, default=str))
    logger.info("Warmed cache for %d users", len(users))

How It Works

1. Cache read — get_or_load checks Redis first. On a hit, it returns the cached value without touching the database.
2. Cache miss — the loader function fetches from the database. The result is written to Redis with a TTL so it auto-expires.
3. Write-through invalidation — after any database write, invalidate deletes the cache entry so the next read loads fresh data.
4. Thundering herd lock — on a cache miss, only one request acquires the lock and loads from the database. Others wait and retry the cache read.
5. Pattern invalidation — invalidate_pattern uses SCAN to find and delete keys matching a glob pattern, useful for invalidating all entries for a user.

Variants

Read-Through with Refresh-Ahead

Refresh the cache before it expires:

def get_or_refresh(key: str, loader: callable, ttl: int = 300) -> dict | None:
    cached = redis_client.get(key)
    if cached:
        result = json.loads(cached)
        # Check if entry is near expiry (within 10% of TTL)
        ttl_remaining = redis_client.ttl(key)
        if ttl_remaining < ttl * 0.1:
            # Refresh in background
            threading.Thread(target=lambda: cache.get_or_load(key, loader, ttl), daemon=True).start()
        return result
    return cache.get_or_load(key, loader, ttl)

Write-Behind (Write-Back)

Write to the cache first, then asynchronously persist to the database:

def write_behind_update(user_id: str, data: dict) -> dict:
    cache_key = f"user:{user_id}"
    redis_client.setex(cache_key, 300, json.dumps(data, default=str))
    redis_client.lpush("write_queue:user", json.dumps({"id": user_id, "data": data}))
    return data

# Background worker processes the queue
def process_write_queue():
    while True:
        item = redis_client.brpop("write_queue:user", timeout=10)
        if item:
            payload = json.loads(item[1])
            db.users.update(payload["id"], payload["data"])

Best Practices

• For a deeper guide, see Complete Guide to Redis Caching Strategies.

• Set a TTL on every cache entry — prevents stale data from persisting indefinitely

• Invalidate on writes — update the database first, then delete the cache entry (not the other way around)

• Handle Redis failures gracefully — the cache is an optimization; the loader should still work without it

• Use short TTLs for mutable data — 30-120 seconds for user data, longer for reference data

Common Mistakes

• Updating cache before database — if the database write fails, the cache has stale data
• Not handling cache misses — if the loader returns None, don’t cache it (or cache it with a short TTL to prevent repeated misses)
• Using KEYS instead of SCAN — KEYS blocks Redis; SCAN is non-blocking and safe for production
• Caching too much — cache only hot data; cold data wastes memory and increases invalidation overhead

Troubleshooting

• Cache and database are out of sync: define a TTL or invalidation policy.
• Hit rate dropped after a deployment: check cache key generation and serialization changes. A new version may use different keys.
• Cold cache causes thundering herd: use cache warming, request coalescing, or single-flight patterns for hot keys.
• Memory usage grows uncontrollably: set max memory policies, eviction thresholds, and key expiration. Audit large values.
• Stale data served to users: implement cache invalidation on write and cache-bust URLs for static assets.

Key Takeaways

• Apply implement the cache-aside pattern with redis when you need a practical solution for your use case.
• Monitor performance after implementation; measure latency, errors, and resource usage before and after.
• Check the Troubleshooting section for common failures; most have documented root causes with fixes.
• Keep dependencies updated and run tests in CI to prevent production regressions.

Common Production Pitfalls

• Copying the example without adapting it to real data volumes and failure modes.
• Skipping load and error-injection tests before the first production deployment.
• Hard-coding values that should be configurable per environment.
• Forgetting to add logging and monitoring at each step.
• Deploying without a rollback plan or a tested backup strategy.
• Assuming the minimal example will scale without adding caching or batching.
• Not documenting the version and configuration used in production.
• Letting the recipe sit unchanged when dependencies or scale evolve.