Caching with Redis
26 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)A product page is read thousands of times and changes once a day. Asking the database every time wastes work. A cache keeps the last answer close by. The hard part is knowing when that answer is no longer true.
Remember answers that rarely change
Cache is a fast, temporary store of answers that were expensive to produce. It can be thrown away at any time, because the real data lives somewhere else, such as the database.
A cache is a fast store of recent answers. Before the slow work, look in the cache. If the answer is there, return it. If not, do the slow work once and store the answer for next time.
cache = {}
def get_product(product_id): if product_id in cache: return cache[product_id] product = load_from_database(product_id) cache[product_id] = product return productThe slow load_from_database runs only on the first request for each product. Later requests are answered from the dictionary.
The cache-aside pattern
Cache-aside is a caching pattern where the application checks the cache first, loads from the database on a miss and fills the cache itself. The cache never talks to the database.
In cache-aside, your code manages the cache itself. A hit returns at once. A miss reads the database, fills the cache, and returns. A change deletes the cached copy so the next read fetches fresh data.
Example 1 · Cache-aside with a database counter
db_reads = 0cache = {}
def load_from_database(product_id): global db_reads db_reads += 1 return {"id": product_id, "name": f"Product {product_id}"}
def get_product(product_id): key = f"product:{product_id}" if key in cache: return cache[key] product = load_from_database(product_id) cache[key] = product return product
print(get_product(1))print(get_product(1))print(get_product(2))print("database reads:", db_reads){'id': 1, 'name': 'Product 1'}
{'id': 1, 'name': 'Product 1'}
{'id': 2, 'name': 'Product 2'}
database reads: 2What this code does
- The second call for product 1 is a hit: the database counter does not grow.
- Keys are strings such as
product:1, the same style Redis uses.
TTL: let old answers expire
TTL is the number of seconds a cached entry may be used. When it passes, the entry expires and the next read is a miss. Redis can remove expired keys for you.
A copy can go out of date. A time to live, or TTL, gives each entry an expiry time. After that time the entry counts as missing and is loaded again. A short TTL means fresher data but more database work.
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return valueExample 2 · A TTL cache with a fake clock
class FakeClock: def __init__(self, now=0.0): self.now = now
def __call__(self): return self.now
def advance(self, seconds): self.now += seconds
class TTLCache: def __init__(self, clock, default_ttl=60): self._clock = clock self._default_ttl = default_ttl self._data = {}
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return value
def set(self, key, value, ttl=None): ttl = self._default_ttl if ttl is None else ttl self._data[key] = (value, self._clock() + ttl)
def delete(self, key): self._data.pop(key, None)
clock = FakeClock()cache = TTLCache(clock, default_ttl=60)
cache.set("product:1", {"name": "Notebook"})print(cache.get("product:1"))
clock.advance(59)print(cache.get("product:1"))
clock.advance(1)print(cache.get("product:1"))
cache.set("rate", 83.2, ttl=5)clock.advance(6)print(cache.get("rate", "expired")){'name': 'Notebook'}
{'name': 'Notebook'}
None
expiredWhat this code does
- At 59 seconds the entry is still valid. At exactly 60 it has expired and the read is a miss.
ttl=5onsetoverrides the default for that key.- The fake clock makes the test run instantly and gives the same result every time.
The cache gets its time from a clock function. In tests we pass a fake clock, so we can move time forward without waiting.
Invalidate when the data changes
Cache invalidation is removing or replacing cached entries when the data behind them changes, so readers do not keep getting an out-of-date answer.
A TTL alone can show old data until it expires. When your own code changes the data, delete the cached key right then. The next read loads the new value.
@app.put("/products/{product_id}")def update_product(product_id: int, body: ProductIn): save_to_database(product_id, body) cache.delete(f"product:{product_id}") return {"ok": True}Example 3 · A cached route that is invalidated on update
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClient
class FakeClock: def __init__(self, now=0.0): self.now = now
def __call__(self): return self.now
def advance(self, seconds): self.now += seconds
class TTLCache: def __init__(self, clock, default_ttl=60): self._clock = clock self._default_ttl = default_ttl self._data = {}
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return value
def set(self, key, value, ttl=None): ttl = self._default_ttl if ttl is None else ttl self._data[key] = (value, self._clock() + ttl)
def delete(self, key): self._data.pop(key, None)
clock = FakeClock()cache = TTLCache(clock, default_ttl=60)db = {1: {"id": 1, "name": "Notebook", "price": 50}}reads = {"db": 0}app = FastAPI()
def get_cache(): return cache
@app.get("/products/{product_id}")def read_product(product_id: int, cache: TTLCache = Depends(get_cache)): key = f"product:{product_id}" cached = cache.get(key) if cached is not None: return cached reads["db"] += 1 if product_id not in db: raise HTTPException(status_code=404, detail="Product not found") cache.set(key, dict(db[product_id])) return db[product_id]
@app.put("/products/{product_id}")def update_price(product_id: int, price: int, cache: TTLCache = Depends(get_cache)): db[product_id]["price"] = price cache.delete(f"product:{product_id}") return db[product_id]
client = TestClient(app)print(client.get("/products/1").json(), reads)print(client.get("/products/1").json(), reads)client.put("/products/1", params={"price": 60})print(client.get("/products/1").json(), reads)print(client.get("/products/9").status_code, reads){'id': 1, 'name': 'Notebook', 'price': 50} {'db': 1}
{'id': 1, 'name': 'Notebook', 'price': 50} {'db': 1}
{'id': 1, 'name': 'Notebook', 'price': 60} {'db': 2}
404 {'db': 3}What this code does
- The cache is a dependency, so tests can give the app a cache with a fake clock.
- After the update, the key is deleted. The next read is a miss and loads price 60.
- The cache stores a copy of the row, not the row itself, so changing the database does not silently change the cache.
- A missing product is not cached; it is a 404 each time.
Example 4 · A decorator that adds caching to any function
class FakeClock: def __init__(self, now=0.0): self.now = now
def __call__(self): return self.now
def advance(self, seconds): self.now += seconds
class TTLCache: def __init__(self, clock, default_ttl=60): self._clock = clock self._default_ttl = default_ttl self._data = {}
def get(self, key, default=None): item = self._data.get(key) if item is None: return default value, expires_at = item if self._clock() >= expires_at: del self._data[key] return default return value
def set(self, key, value, ttl=None): ttl = self._default_ttl if ttl is None else ttl self._data[key] = (value, self._clock() + ttl)
def delete(self, key): self._data.pop(key, None)
MISSING = object()stats = {"hits": 0, "misses": 0}clock = FakeClock()cache = TTLCache(clock, default_ttl=30)
def cached(ttl): def wrap(func): def inner(*args): key = f"{func.__name__}:{args}" value = cache.get(key, MISSING) if value is not MISSING: stats["hits"] += 1 return value stats["misses"] += 1 value = func(*args) cache.set(key, value, ttl) return value return inner return wrap
@cached(ttl=10)def category_total(category): return {"books": 120, "pens": 40}.get(category)
print(category_total("books"), category_total("books"), category_total("toys"), category_total("toys"))clock.advance(11)print(category_total("books"))print(stats)120 120 None None
120
{'hits': 2, 'misses': 3}What this code does
- A sentinel
MISSINGtells a storedNoneapart from a missing key, so evenNoneanswers are cached. - After 11 seconds the entries have expired, so
booksis a miss again.
Delete the key after the database change succeeds. Do not try to update the cached copy: deleting is simpler and cannot leave a wrong value behind.
Where Redis fits
Redis is a fast in-memory data store used as a shared cache, a counter store and a message broker. Apps talk to it over the network with a client library such as redis-py.
An in-process dictionary is lost on restart, and each app copy has its own. Redis is a separate in-memory server that all copies share, with a TTL on each key.
The next example is redis-py code. It needs a Redis server and the redis package, so it is a sample and is not run here.
Example 5 · The same pattern with redis-py (sample, not run here)
import json
import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
def load_from_database(product_id): return {"id": product_id, "name": "Notebook", "price": 50}
def get_product(product_id): key = f"product:{product_id}" cached = r.get(key) if cached is not None: return json.loads(cached) product = load_from_database(product_id) r.set(key, json.dumps(product), ex=60) return product
def update_price(product_id, price): save_price(product_id, price) r.delete(f"product:{product_id}")
print(get_product(1))print(r.ttl("product:1")){'id': 1, 'name': 'Notebook', 'price': 50}
60What this code does
- Needs
pip install redisand a Redis server onlocalhost:6379. It is shown, not run, in this course. - Redis stores text, so the dictionary is saved with
json.dumpsand read back withjson.loads. ex=60is the TTL in seconds.r.ttl(key)shows how many seconds are left.
Next step
Check your understanding
12 quick questions