This post was originally published in Portuguese on my blog: Cache-aside com FastAPI e Redis: invalidação e requisições simultâneas.

Anyone who has put an API in production knows the story: the read endpoint is simple, but the query behind it is expensive, and the same row gets requested hundreds of times per minute. Then someone suggests "put a Redis in front of it". It works, but the hard part isn't writing to the cache. It's deciding when to read, when to invalidate, and what to do when ten requests arrive together and the cache is empty.

In this post I built a small example with FastAPI and Redis using the cache-aside pattern, paying attention to those two annoying points: invalidation and concurrent requests. Everything runs with docker compose, and the tests don't need a real Redis.

The idea: cache-aside

In cache-aside, the application is in charge, not the cache. The read flow is always the same: ask Redis; if it has the value, return it; if not, fetch from the source of truth, store it in Redis with a TTL and return it. On writes, we change the source and delete the cache key, letting the next read repopulate it. It's less clever than write-through, but it's easy to understand and easy to debug.

In the example, the "database" is a repository that uses asyncio.sleep to simulate latency and keeps a query counter. That counter is what lets the tests prove the cache is working, instead of trusting a gut feeling.

The cached read

The read endpoint is practically the pattern described in code. Notice that the response reports the origem (origin) of the data, which helps a lot when testing and when watching the behavior with curl.

em_cache = await r.get(chave(produto_id)) if em_cache: return {"origem": "cache", "produto": json.loads(em_cache)} produto = await repo.buscar(produto_id) if produto is None: raise HTTPException(status_code=404, detail="Produto não encontrado") await r.set(chave(produto_id), json.dumps(produto), ex=TTL_SEGUNDOS) return {"origem": "banco", "produto": produto}

One detail: a nonexistent product returns 404 and is not cached. You can cache the absence too (negative caching), but then you need a short TTL and care with invalidation when the item starts to exist. For an example, I left it out.

The concurrent requests problem

This code has a classic flaw. If the key expires and ten requests arrive at the same time, all ten see an empty cache and all ten hit the database. That's a cache stampede, and it tends to happen at the worst possible moment, under heavy load.

A simple solution for a single process is a per-key lock. The first request takes the lock and goes to the database; the others wait, and when they get in they check the cache again (the famous double-check), finding the value the first one already stored.

lock = request.app.state.locks.setdefault(produto_id, asyncio.Lock()) async with lock: em_cache = await r.get(chave(produto_id)) if em_cache: return {"origem": "cache", "produto": json.loads(em_cache)} produto = await repo.buscar(produto_id) ...

This lock lives in the process memory. With several API instances, each one would make at most one query per key, which already reduces the damage a lot, but doesn't eliminate it. If you need a guarantee across instances, the way to go is a distributed lock in Redis itself (SET key value NX EX), with the expiration caveats that come with it.

Invalidation on write

When the product changes, we delete the key. That's it. The next read goes to the database, gets the new data and repopulates the cache. The TTL still exists as a safety net: if some write forgets to invalidate, the stale data goes away by itself within 30 seconds.

@app.put("/produtos/{produto_id}") async def atualizar(produto_id: int, dados: dict, request: Request): PRODUTOS[produto_id] = {**PRODUTOS[produto_id], **dados, "id": produto_id} await request.app.state.redis.delete(chave(produto_id)) return PRODUTOS[produto_id]

Testing without Redis

The tests use fakeredis, which implements the client's async API in memory. Since create_app receives the client and the repository as parameters, each test builds its own app, with its own counter, and nothing depends on execution order. The most interesting test fires ten parallel requests and checks that there was a single database query:

def test_requisicoes_simultaneas_geram_uma_unica_consulta(): async def cenario(): app, repo = novo_app(latencia=0.1) async with cliente_http(app) as c: respostas = await asyncio.gather(*[c.get("/produtos/1") for _ in range(10)]) return [r.json()["origem"] for r in respostas], repo.consultas origens, consultas = rodar(cenario()) assert consultas == 1 assert origens.count("banco") == 1 assert origens.count("cache") == 9

The other tests cover the first read coming from the database and the second from the cache, the 404, invalidation after the PUT, and the TTL set on the key.

Running with Docker

The compose file brings up the API and a Redis with a healthcheck, and the API only starts after Redis answers the ping. The tests live in a separate service behind a profile, so they don't start together with the application.

services: redis: image: redis:7-alpine healthcheck: test: ["CMD", "redis-cli", "ping"] app: build: { context: ., target: runtime } environment: REDIS_URL: redis://redis:6379/0 CACHE_TTL: "30" ports: ["8000:8000"] depends_on: redis: condition: service_healthy 
docker compose up --build curl localhost:8000/produtos/1 # origem: banco curl localhost:8000/produtos/1 # origem: cache docker compose run --rm tests

If you repeat the curl after 30 seconds, you'll see the origin go back to banco, because the TTL expired. And if you do a PUT on the product, the next read also goes back to the database.

Code

The complete project, with Dockerfile, compose, tests and a CI workflow, is at programero-python/2026-11-cache-aside-fastapi-redis. The natural next step is to swap the in-memory lock for a distributed one and measure the difference with several instances.