How to implement stale-while-revalidate caching in Python
A Python cache wrapper that returns a stale cached value with a fallback flag when the upstream fetch fails, using TTL-based freshness checks.
Python code
42 linesimport time
from functools import lru_cache
class CachedService:
def __init__(self, fetch_func, ttl=5):
self.fetch_func = fetch_func
self.ttl = ttl
self._cache = {}
self._timestamp = {}
def get(self, key):
now = time.time()
if key in self._cache and now - self._timestamp[key] < self.ttl:
return self._cache[key], False
try:
value = self.fetch_func(key)
if value is None:
raise ValueError("fetch returned None")
except Exception:
if key in self._cache:
return self._cache[key], True
raise
self._cache[key] = value
self._timestamp[key] = now
return value, False
if __name__ == "__main__":
calls = []
def flaky_fetch(key):
calls.append(key)
if len(calls) < 3:
raise ConnectionError("temporary failure")
return f"fresh-{key}"
svc = CachedService(flaky_fetch, ttl=10)
print(svc.get("a")) # raises
print(svc.get("a")) # raises
print(svc.get("a")) # fresh result, cached
print(svc.get("a")) # cache hit, not stale
Output
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 19, in get
ConnectionError: temporary failure
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 19, in get
ConnectionError: temporary failure
('fresh-a', False)
('fresh-a', False)
How it works
The CachedService keeps two dicts: one for values and one for fetch timestamps. On each get, it checks the TTL window; if the entry is still fresh, it returns immediately. If the entry is stale or missing, it attempts the fetch function. On success, it updates both dicts and returns a (value, stale=False) tuple. On failure, it falls back to whatever stale value exists, returning (value, stale=True); if no stale value exists, the original exception propagates. This is the classic stale-while-revalidate pattern in ~40 lines of stdlib-only code.
Common mistakes
- Treating `lru_cache` as the primary cache when you need TTL-based invalidation (it has no expiry support).
- Swallowing exceptions for keys that have no stale value, which hides real infrastructure failures.
- Not distinguishing a stale read from a fresh one in the caller, so downstream code can't react to degraded data.
- Updating the timestamp before the fetch completes, which shortens the effective TTL on failure.
Variations
- Use `asyncio.Lock` to deduplicate concurrent fetches for the same key.
- Wrap the fetch in a retry with exponential backoff before falling back to stale data.
Real-world use cases
- Serving cached API responses to users while a backend microservice is briefly down or slow.
- Returning previously rendered config or feature flags when a config service is unreachable during rollouts.
- Keeping last-known-good model predictions available when an ML inference service fails over.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.