🗃️ Caching & Memoization

The Cheapest Win Under Load

If the same input produces the same output, computing it twice is waste. At 10,000 req/sec, shaving a 50ms DB round-trip off 90% of requests via a cache is often the difference between one box and twenty.

functools.lru_cache / functools.cache

For pure functions, memoization is one decorator. lru_cache(maxsize=N) keeps the N most recently used results and evicts the least recently used; functools.cache (3.9+) is just lru_cache(maxsize=None) — unbounded, so only use it when the key space is small and fixed.

from functools import lru_cache, cache

@lru_cache(maxsize=1024)          # bounded — safe for a large/unbounded key space
def expensive(n: int) -> int:
    # pretend this is a heavy pure computation or a slow lookup
    return sum(i * i for i in range(n))

@cache                            # unbounded — fine ONLY for small fixed key sets
def config_for(env: str) -> dict:
    return _load_config(env)      # e.g. "dev"/"staging"/"prod"

expensive(10_000)                 # computed
expensive(10_000)                 # instant — served from cache

print(expensive.cache_info())     # CacheInfo(hits=1, misses=1, maxsize=1024, currsize=1)
expensive.cache_clear()           # wipe it (e.g. on config reload)

A Simple TTL Cache

lru_cache has no notion of time — entries live until evicted. When data goes stale after N seconds, you want a TTL cache: store (value, expiry) and treat expired entries as misses.

import time

class TTLCache:
    def __init__(self, ttl: float):
        self.ttl = ttl
        self._store: dict = {}     # key -> (value, expires_at)

    def get(self, key):
        hit = self._store.get(key)
        if hit is None:
            return None
        value, expires_at = hit
        if time.monotonic() >= expires_at:
            self._store.pop(key, None)   # lazily evict on read
            return None
        return value

    def set(self, key, value):
        self._store[key] = (value, time.monotonic() + self.ttl)

# In production, reach for cachetools — it handles eviction + max size + TTL for you:
from cachetools import TTLCache
cache = TTLCache(maxsize=10_000, ttl=60)   # bounded AND time-limited
cache["user:42"] = profile                 # expires 60s after write

The Big High-Load Trap: Cache Stampede

In Threading — a per-key lock, double-checked

Keep a dict of threading.Lock keyed by cache key. A thread that misses grabs the key's lock, re-checks the cache (someone may have filled it while it waited), and only then recomputes. The rest block on the lock and get the freshly-cached value for free.

import threading

class SingleFlightCache:
    def __init__(self, compute, ttl: float):
        self._compute = compute
        self._cache = TTLCache(ttl)              # from earlier
        self._locks: dict = {}
        self._locks_guard = threading.Lock()     # guards the dict of locks itself

    def _lock_for(self, key) -> threading.Lock:
        with self._locks_guard:
            lock = self._locks.get(key)
            if lock is None:
                lock = self._locks[key] = threading.Lock()
            return lock

    def get(self, key):
        value = self._cache.get(key)
        if value is not None:
            return value                          # fast path: hit, no locking

        with self._lock_for(key):                 # only contenders for THIS key serialize
            value = self._cache.get(key)          # double-check: filled while we waited?
            if value is not None:
                return value
            value = self._compute(key)            # exactly ONE thread runs this
            self._cache.set(key, value)
            return value

In Asyncio — a single-flight coalescer

Same idea, but instead of a lock you cache the in-flight coroutine's Task. The first caller for a key creates the task; every subsequent caller awaits the same task and shares its result. No thread locks — just structured sharing of one Future.

import asyncio

class AsyncSingleFlight:
    def __init__(self, compute, ttl: float):
        self._compute = compute                   # async callable
        self._cache = TTLCache(ttl)
        self._inflight: dict = {}                 # key -> asyncio.Task

    async def get(self, key):
        value = self._cache.get(key)
        if value is not None:
            return value                          # fast path: hit

        task = self._inflight.get(key)
        if task is None:
            # first caller for this key creates the ONE task everyone shares
            task = asyncio.ensure_future(self._fill(key))
            self._inflight[key] = task
        return await task                         # all concurrent callers await the same task

    async def _fill(self, key):
        try:
            value = await self._compute(key)      # runs exactly once per key
            self._cache.set(key, value)
            return value
        finally:
            self._inflight.pop(key, None)         # let the next miss start fresh

# 1,000 concurrent requests for the same cold key → ONE actual DB call:
sf = AsyncSingleFlight(compute=fetch_user_from_db, ttl=60)
results = await asyncio.gather(*(sf.get("user:42") for _ in range(1000)))
# all 1,000 resolve from a single fetch_user_from_db("user:42")

Multiprocessing & Multi-Instance — local caches don't share

lru_cache and any in-process dict live in one interpreter's memory. With 4 worker processes (or 20 pods), you have 4 (or 20) independent caches: the hit rate drops, and a single-flight lock in process A does nothing to stop process B from also stampeding the DB.

# Shared cache across processes / machines: Redis (or memcached).
import redis, json

r = redis.Redis()

def get_user(user_id: str) -> dict:
    key = f"user:{user_id}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)                 # shared hit across ALL processes/pods
    value = fetch_user_from_db(user_id)
    r.set(key, json.dumps(value), ex=60)          # ex = TTL in seconds
    return value

Two More Patterns Worth Knowing

• Negative caching — cache "not found" / errors too (with a shorter TTL). Otherwise a flood of requests for a nonexistent key misses every time and stampedes the DB just as hard as a hot key would. Guards against cache-penetration attacks.
• Stale-while-revalidate — on expiry, serve the stale value immediately and trigger a background refresh. Nobody waits on a recompute, and the stampede window disappears entirely because reads never block on a miss.

# Stale-while-revalidate sketch: keep value + a "soft" and "hard" expiry.
def get_swr(key):
    value, soft, hard = _cache.get3(key)          # (value, soft_expiry, hard_expiry)
    now = time.monotonic()
    if value is not None and now < hard:
        if now >= soft:
            schedule_background_refresh(key)       # refresh async; don't block the caller
        return value                               # serve immediately, possibly stale
    return recompute_and_cache(key)                # only truly-expired hits pay the cost

Key Takeaways