Cache-aside com FastAPI e Redis: invalidação e requisições simultâneas

Todo mundo que já colocou uma API em produção conhece o filme: o endpoint de leitura é simples, mas a consulta por trás dele é cara, e a mesma linha é pedida centenas de vezes por minuto. Aí alguém sugere "põe um Redis na frente". Funciona, mas o difícil não é gravar no cache, é decidir quando ler, quando invalidar e o que fazer quando dez requisições chegam juntas e o cache está vazio.

Neste post eu montei um exemplo pequeno com FastAPI e Redis usando o padrão cache-aside, e dei atenção a esses dois pontos chatos: invalidação e requisições simultâneas. Tudo roda com docker compose e os testes não precisam de um Redis de verdade.

A ideia: cache-aside

No cache-aside quem manda é a aplicação, não o cache. O fluxo de leitura é sempre o mesmo: pergunta ao Redis; se tem, devolve; se não tem, busca na fonte de verdade, grava no Redis com um TTL e devolve. Na escrita, a gente muda a fonte e apaga a chave do cache, deixando a próxima leitura repovoar. É menos esperto que write-through, mas é fácil de entender e de depurar.

No exemplo, o "banco" é um repositório com asyncio.sleep simulando latência e um contador de consultas. Esse contador é o que permite provar nos testes que o cache está funcionando, em vez de confiar no olhômetro.

A leitura com cache

O endpoint de leitura é praticamente a descrição do padrão em código. Repare que a resposta informa a origem do dado, o que ajuda muito na hora de testar e de ver o comportamento com o 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}

Um detalhe: produto inexistente retorna 404 e não é cacheado. Dá para cachear a ausência também (cache negativo), mas aí você precisa de um TTL curto e de cuidado com invalidação quando o item passar a existir. Para um exemplo, fica de fora.

O problema das requisições simultâneas

Esse código tem uma falha clássica. Se a chave expira e chegam dez requisições ao mesmo tempo, as dez veem o cache vazio e as dez vão ao banco. Isso é o cache stampede, e costuma acontecer justamente no pior momento, com a carga alta.

Uma solução simples para um processo só é uma trava por chave. A primeira requisição pega o lock e vai ao banco; as outras esperam, e quando entram conferem o cache de novo (o famoso double-check), encontrando o valor que a primeira já gravou.

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)
    ...

Esse lock vive na memória do processo. Com várias instâncias da API, cada uma faria no máximo uma consulta por chave, o que já reduz muito o estrago, mas não elimina. Se precisar de garantia entre instâncias, o caminho é um lock distribuído no próprio Redis (SET chave valor NX EX), com os cuidados de expiração que isso traz.

Invalidação na escrita

Quando o produto muda, apagamos a chave. Só isso. A próxima leitura vai ao banco, pega o dado novo e repovoa o cache. O TTL continua existindo como rede de segurança: se alguma escrita esquecer de invalidar, o dado errado some sozinho em até 30 segundos.

@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]

Testando sem Redis

Os testes usam fakeredis, que implementa a API assíncrona do cliente em memória. Como o create_app recebe o cliente e o repositório por parâmetro, cada teste monta o seu próprio app, com o seu próprio contador, e nada depende da ordem de execução. O teste mais interessante dispara dez requisições em paralelo e confere que houve uma única consulta ao banco:

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

Os outros testes cobrem a primeira leitura vindo do banco e a segunda do cache, o 404, a invalidação depois do PUT e o TTL definido na chave.

Rodando com Docker

O compose sobe a API e um Redis com healthcheck, e a API só inicia depois que o Redis responde ao ping. Os testes ficam em um serviço separado, atrás de um profile, para não subirem junto com a aplicação.

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

Se você repetir o curl depois de 30 segundos, vai ver a origem voltar para banco, porque o TTL expirou. E se fizer um PUT no produto, a próxima leitura também volta ao banco.

Código

O projeto completo, com Dockerfile, compose, testes e um workflow de CI, está em programero-python/2026-11-cache-aside-fastapi-redis. No próximo passo natural, vale trocar o lock em memória por um lock distribuído e medir a diferença com várias instâncias.