Caching — Redis for Accelerating Hot Queries
A cache is like a refrigerator—you keep the ingredients you use most often (hot data) in the fridge so they're always within reach, and you don't have to go to the supermarket (database) every time. But ingredients expire (TTL), and when you restock, you need to update them (cache invalidation).
1. What You'll Learn
- Redis Connection Management:
redis-pyIntegrating an Asynchronous Client with FastAPI - Caching Strategies: Cache-Aside Pattern, TTL Expiration, Cache Key Naming Design
- Cache Invalidation: Proactive invalidation during write operations; cache consistency during batch price updates
- Cache Penetration/Breakthrough/Avalanche Protection: Bloom Filters, Mutual Exclusion Locks, Random TTL
- Alice Scenario: Caching Prices of Popular Products—P99 latency reduced from 200 ms to 5 ms
2. Alice's True Story
(1) Pain Point: Popular Queries Overwhelm the Database
PriceTracker has 100 popular products that account for 80% of query volume, with 8,000 queries per second. Every query goes through PostgreSQL, causing the database CPU to spike to 90% and resulting in a P99 latency of 200 ms. Charlie says adding a database instance would cost USD 500 per month, but 80% of the query results won't change within 5 minutes.
(2) Solutions for Redis Caching
Cache price data for popular products in Redis. For queries, check Redis first (5 ms); if there's no hit, query PostgreSQL (50 ms) and write the result back to Redis with a 5-minute TTL.
PYTHON
async def get_product_cached(product_id: int, db, redis):
cached = await redis.get(f"product:{product_id}")
if cached:
return json.loads(cached)
product = await db.execute(select(Product).where(Product.id == product_id))
data = product.scalar_one_or_none()
await redis.setex(f"product:{product_id}", 300, json.dumps(data))
return data
(3) Revenue
Top queries hit the Redis cache 80% of the time; P99 latency dropped from 200 ms to 5 ms; database QPS dropped from 8,000 to 1,600; Charlie's database instances were reduced from 4 to 1; saving USD 1,500 per month.
3. Redis Connection Management
(1) Asynchronous Client Integration
(1) ▶ Example: Redis Connection Dependencies
PYTHON
import redis.asyncio as aioredis
from fastapi import FastAPI, Depends
REDIS_URL = "redis://localhost:6379/0"
# Application-level Redis connection pool
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)
async def get_redis() -> aioredis.Redis:
"""Request-scoped Redis client from connection pool"""
client = aioredis.Redis(connection_pool=redis_pool)
try:
yield client
finally:
await client.aclose()
# Startup/shutdown lifecycle
app = FastAPI()
@app.on_event("startup")
async def startup():
global redis_pool
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)
@app.on_event("shutdown")
async def shutdown():
await redis_pool.disconnect()
Output:
TEXT
# Function defined successfully
(2) Selecting Redis Data Structures
| Structure | Command | Use Case | PriceTracker Use Case |
|---|---|---|---|
| String | SET/GET | Single-Value Cache | Product Price Cache |
| Hash | HSET/HGET | Object Cache | Product Detail Cache |
| List | LPUSH/LRANGE | Time Series | Price Change Log |
| Set | SADD/SMEMBERS | Deduped Set | Pushed Price IDs |
| ZSet | ZADD/ZRANGE | Leaderboard | Top-Selling Products |
4. Cache-Aside Mode
(1) Read/Write Process
flowchart TD
Request[Read Request] --> CheckRedis{Cache Hit?}
CheckRedis -->|Yes| Return[Return Cached Data]
CheckRedis -->|No| QueryDB[Query Database]
QueryDB --> WriteCache[Write to Redis with TTL]
WriteCache --> Return2[Return Data]
WriteRequest[Write Request] --> UpdateDB[Update Database]
UpdateDB --> Invalidate[Invalidate Cache]
Invalidate --> Return3[Return Success]
(1) ▶ Example: Cache-Aside Product Query
PYTHON
import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis
app = FastAPI()
@app.get("/api/v1/products/{product_id}")
async def get_product(
product_id: int,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
):
# Step 1: Check Redis cache
cache_key = f"product:{product_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# Step 2: Cache miss - query database
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Step 3: Write to cache with TTL
product_data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(product_data)) # 5 min TTL
return product_data
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Proactive invalidation during write operations
PYTHON
@app.put("/api/v1/products/{product_id}")
async def update_product(
product_id: int,
update: ProductUpdate,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
user=Depends(get_current_user),
):
repo = ProductRepository(db)
product = await repo.update(product_id, update.model_dump(exclude_unset=True))
# Invalidate cache after write
await redis.delete(f"product:{product_id}")
return ProductResponse.model_validate(product).model_dump()
Output:
TEXT
# Function defined successfully
(2) Cache Key Naming Conventions
| Mode | Element Name | TTL | Description |
|---|---|---|---|
| In-Memory Cache | product:{id} |
5 min | Single Product Details |
| List Cache | products:cat:{category}:p:{page} |
2 min | Paginated List |
| Counter | count:products:cat:{category} |
1 min | Product Count |
| Top Picks | ranking:products:hot |
10 min | ZSet Rankings |
| Rate Limiting Count | ratelimit:{ip} |
60 s | API Rate Limiting |
5. Cache Penetration/Bypass/Avalanche Protection
(1) Three Major Caching Issues
| Issue | Cause | Mitigation Strategy |
|---|---|---|
| Penetration | Checking for Non-Existent Data, Bypassing the Cache to Access the DB Directly | Bloom Filters, Caching Null Values |
| Overload | A massive surge of requests hits the database the moment a hot key expires | Mutual exclusion locks, never-expiring keys + asynchronous updates |
| Avalanche | Mass Key Expiration | Random TTL Offset |
(1) ▶ Example: Caching Null Values to Prevent Penetration
PYTHON
@app.get("/api/v1/products/{product_id}")
async def get_product_with_null_cache(
product_id: int,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
):
cache_key = f"product:{product_id}"
cached = await redis.get(cache_key)
if cached:
data = json.loads(cached)
if data.get("_null"):
raise HTTPException(status_code=404, detail="Product not found")
return data
# Query DB
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if not product:
# Cache null value with short TTL to prevent cache penetration
await redis.setex(cache_key, 60, json.dumps({"_null": True}))
raise HTTPException(status_code=404, detail="Product not found")
product_data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(product_data))
return product_data
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Preventing Breakthrough in Mutual Exclusion Locks
PYTHON
import asyncio
async def get_product_with_lock(
product_id: int,
db: AsyncSession,
redis: Redis,
):
cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# Try to acquire lock (only one request rebuilds cache)
lock_acquired = await redis.set(lock_key, "1", nx=True, ex=10)
if lock_acquired:
try:
# This request rebuilds the cache
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if product:
data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(data))
return data
finally:
await redis.delete(lock_key)
else:
# Other requests wait briefly and retry cache
await asyncio.sleep(0.1)
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
raise HTTPException(status_code=404, detail="Product not found")
Output:
TEXT
# Function defined successfully
(3) ▶ Example: Random TTL to Prevent Avalanches
PYTHON
import random
def get_cache_ttl(base_ttl: int = 300, jitter: int = 60) -> int:
"""Add random jitter to TTL to prevent cache avalanche"""
return base_ttl + random.randint(-jitter, jitter)
# Usage
await redis.setex(cache_key, get_cache_ttl(300, 60), json.dumps(data))
# TTL: 240-360 seconds (5 min ± 1 min jitter)
Output:
TEXT
# Function defined successfully
(2) Comparison of Cache Coherency Policies
| Strategy | Write Operation | Advantages | Disadvantages |
|---|---|---|---|
| Cache-Aside | Update the DB first, then clear the cache | Simple and reliable | Brief window of inconsistency |
| Write-Through | Writes to both cache and DB simultaneously | High consistency | High write latency |
| Write-Behind | Writes to cache first, then asynchronously to the database | Good write performance | Potential data loss |
❓ FAQ
Q Should the cache store data in JSON or Pickle?
A We recommend JSON. It's human-readable, cross-language, and secure (Pickle has deserialization vulnerabilities). orjson is 3-10 times faster than json.
Q What is the appropriate TTL setting?
A It depends on how frequently the data changes. PriceTracker price data is updated every 5 minutes (brief inconsistencies are allowed), product details every 30 minutes, and leaderboards every 10 minutes. A random offset is added to prevent an avalanche.
Q How should I set the Redis connection pool size?
A Match it to the number of FastAPI Workers. Each Worker requires 5-10 Redis connections. 4 Workers x 10 = 40 connections; set
max_connections to 50 to allow for some headroom.Q How is a Bloom filter implemented?
A Using the
redisbloom module or the Python pybloom_live library. All existing product IDs are added to the Bloom filter, and a check is performed before each query. The false positive rate is approximately 1%, which is acceptable.Q What should I do if there is an inconsistency between the cache and the database?
A In Cache-Aside mode, temporary inconsistencies are normal. Strategy: Update the DB first, then delete the cache (rather than updating the cache), and use a delayed double-delete (delete cache → update DB → delay before deleting again) to minimize the inconsistency window.
Q How do I monitor the cache hit rate?
A Redis's
INFO stats command returns keyspace_hits and keyspace_misses; the hit rate is calculated as hits / (hits + misses). Integrate this into a Prometheus + Grafana monitoring dashboard.📖 Summary
- Redis Asynchronous Client (
redis.asyncio) + Connection Pool +yieldDependency Injection Integration with FastAPI - Cache-Aside mode: For reads, check the cache first; if there is no hit, query the database and write the data to the cache; for writes, update the database and delete the cache entry.
- Cache keys are named according to the
{entity}:{id}pattern, with different TTLs set for different scenarios - Null-value caching to prevent penetration, mutual exclusion locks to prevent breakthroughs, and random TTL to prevent avalanches—three major defense strategies
- PriceTracker: Cache hit rate for popular products is 80%; P99 latency has dropped from 200 ms to 5 ms
📝 Exercises
- Basic Exercise (Difficulty ⭐): Configure the Redis asynchronous client dependency, implement simple GET/SET caching endpoints, and verify that data can be stored and retrieved. Hint:
redis.asyncio+Depends(get_redis)+setex() - Advanced Problem (Difficulty ⭐⭐): Implement the Cache-Aside pattern for the PriceTracker product lookup: Read from cache → If cache miss, query the DB → Write to cache (TTL 5 minutes); after the write operation updates the DB, delete the cache entry. Hint:
await redis.get(key)+await redis.delete(key) - Challenge Question (Difficulty ⭐⭐⭐): Implement comprehensive cache protection—empty-value cache penetration prevention (short TTL of 60s), mutex lock breakthrough prevention (
SET lock NX EX), and random TTL avalanche prevention (300s ± 60s)—and write tests to verify that the protection logic is correct. Hint:redis.set(lock_key, "1", nx=True, ex=10)+random.randint()
---|
