We already have a baseline and know one of task-tracker-api's weak points from Part 3. Now for the first optimization: adding a caching layer with Redis, so repeated read requests don't always hit the database.
Why Read Endpoints Are a Good Fit for Caching
GET /api/tasks and GET /api/tasks/{id} have exactly the right characteristics for caching: read far more often than they're changed. If a hundred people open the same task list within a minute, there's no strong reason to query the database a hundred times — the data is (almost certainly) identical. This is the most common pattern that benefits massively from a cache.
The Cache-Aside Pattern
The caching pattern we use is called cache-aside (sometimes called lazy loading): the application itself decides when to read from the cache, when to read from the database, and when to populate the cache — rather than the database or cache staying automatically in sync on their own. The flow:
1. Request comes in, check the cache first
2. In the cache (cache HIT)? → return it immediately, DONE, never touch the database
3. Not in the cache (cache MISS)? → query the database → store the result in the cache → return it
@router.get("", response_model=List[TaskRead])
def list_tasks(
done: Optional[bool] = None,
session: Session = Depends(get_session),
cache: redis.Redis = Depends(get_redis),
):
key = list_cache_key(done)
cached = cache.get(key)
if cached is not None:
return json.loads(cached) # cache HIT — never touches the database
query = select(Task)
if done is not None:
query = query.where(Task.done == done)
results = session.exec(query).all()
data = [TaskRead.model_validate(task).model_dump(mode="json") for task in results]
cache.set(key, json.dumps(data), ex=settings.cache_ttl_seconds) # populate cache for next time
return data
Notice cache: redis.Redis = Depends(get_redis) — the exact same pattern as Depends(get_session) for the database. This is why the dependency injection we learned in the FastAPI series is so useful: adding a new dependency (the cache) to an existing endpoint doesn't change how that endpoint is called at all — it just adds one parameter.
TTL — Why a Cache Should Never Live Forever
TTL (Time To Live) is the duration before cached data is automatically considered expired and removed. In our code, this is set via the ex parameter (in seconds) when calling cache.set(...), controlled by settings.cache_ttl_seconds (default 30 seconds).
Why not cache forever? Because cached data can become stale (out of date / out of sync with the database) the moment something changes. TTL is a simple safety net: even if our invalidation mechanism has a bug, or data changes through some other path (say, a manual database query), stale data will live for at most the TTL duration, then automatically "heal itself" once the cache expires and gets refilled from the database.
Cache Invalidation — The Hardest Part of Caching
There's a famous saying in software engineering: "There are only two hard things in Computer Science: cache invalidation and naming things." TTL alone isn't enough for our case — imagine a user creates a new task, but the list endpoint still shows the old cached version for the next 30 seconds. That's a bug users genuinely notice.
The fix: the moment a write happens (create/update/delete), we explicitly delete the relevant cache entries, so the next request automatically pulls fresh data from the database:
def invalidate_task_cache(client, task_id: Optional[int] = None) -> None:
if task_id is not None:
client.delete(task_cache_key(task_id))
client.delete(*LIST_CACHE_KEYS)
@router.patch("/{task_id}", response_model=TaskRead)
def update_task(task_id: int, payload: TaskUpdate, session: Session = Depends(get_session), cache: redis.Redis = Depends(get_redis)):
...
session.commit()
session.refresh(task)
invalidate_task_cache(cache, task_id) # delete this task's detail cache AND every list cache
return task
Notice we delete TWO kinds of cache entries on update: the task's own detail cache (task:{id}), AND every list cache (tasks:list:*) — because a changed task could affect a list's results (e.g. the ?done=true filter). For an app this small, wiping every list key at once is simple and cheap enough. In a much larger app with complex filter patterns, the invalidation strategy can get more sophisticated (granular per-filter invalidation, or cache tags) — but the underlying principle stays the same.
Measuring the Impact
Now it's time to prove this caching actually helps, not just "looks fancier." Run the app (this time it needs Postgres and Redis — start them with docker compose up -d postgres redis), seed a fair amount of data (we use 5,000 tasks so the query is heavy enough to show a difference), then compare performance with a cold cache (just flushed, almost every request hits the database) vs. a warm cache (immediately repeated, almost every request hits Redis):
docker exec <redis-container> redis-cli FLUSHALL
python benchmark/load_test.py --url http://localhost:8000/api/tasks --requests 100 --concurrency 10
# repeat the exact same command, WITHOUT flushing again:
python benchmark/load_test.py --url http://localhost:8000/api/tasks --requests 100 --concurrency 10
The result, on the same machine as Part 3:
| Cold cache | Warm cache | |
|---|---|---|
| Throughput | 18.4 req/s | 24.9 req/s |
| Average latency | 529.5 ms | 389.7 ms |
| Median latency | 390.2 ms | 388.2 ms |
| P95 latency | 1847.0 ms | 524.9 ms |
| Max latency | 1903.4 ms | 605.7 ms |