Especialização em Backend com Python & FastAPI • FastAPI & Python 3.12+ • Pydantic v2, SQLModel, Async/Await e Microsserviços

🗺️ Mapa Conceitual do Tópico

flowchart TD
    A["Requisição do Cliente: GET /produtos/42"] --> B["FastAPI Handler"]
    B --> C["redis.get('produto:42')"]
    C --> D{"Existe no Cache (Hit)?"}
    
    D -->|"✅ Cache HIT"| E["Deserializa JSON e Retorna Imediatamente (< 1ms)"]
    
    D -->|"❌ Cache MISS"| F{"Lock Distribuído Adquirido? (NX)"}
    F -->|"Sim"| G["Consulta Banco Relacional (PostgreSQL)"]
    F -->|"Não (Aguardando)"| H["Aguarda 50ms e Tenta Novamente o Cache"]
    H --> C
    
    G --> I["Grava no Redis com TTL + Jitter"]
    I --> J["Libera Lock Distribuído"]
    J --> K["Retorna Resposta ao Cliente"]

    subgraph MUTACAO["Invalidação Ativa na Escrita"]
        L["PUT /produtos/42"] --> M["Atualiza Banco Relacional"]
        M --> N["redis.delete('produto:42')"]
        N --> O["Retorna Confirmação"]
    end

    style A fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style B fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    style C fill:#ede7f6,stroke:#7e57c2,stroke-width:2px
    style D fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style E fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px
    style F fill:#fffde7,stroke:#fbc02d,stroke-width:2px
    style G fill:#ffebee,stroke:#e53935,stroke-width:2px
    style I fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style K fill:#e0f2f1,stroke:#009688,stroke-width:2px
    style L fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style M fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    style N fill:#ffebee,stroke:#e53935,stroke-width:2px
    style O fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px

🏛️ 1. O Papel do Cache Distribuído na Arquitetura FastAPI

Em aplicações corporativas, grande parte das consultas a bancos relacionais é redundante: listas de produtos em destaque, configurações de sistema, dados de perfil de usuário ou tabelas de frete são lidas centenas de vezes por segundo enquanto sofrem poucas alterações ao longo do dia. Cada consulta que atinge o PostgreSQL consome recursos de CPU, I/O de disco e conexões limitadas do pool.

O Redis é uma estrutura de dados em memória de altíssimo desempenho baseada em C, capaz de entregar tempos de resposta abaixo de 0.5 milissegundos. Diferente de um cache local na memória do processo (como um dicionário Python ou functools.lru_cache), o Redis atua como um Cache Distribuído:

  • Se sua API estiver escalada com 10 réplicas em contêineres Docker, todas as réplicas compartilham exatamente a mesma base de cache.
  • O reinício de uma réplica do FastAPI não limpa o cache.
  • A invalidação de um dado por uma réplica é refletida instantaneamente para todas as outras réplicas da frota.

⚙️ 2. Gerenciamento do Pool Assíncrono com lifespan

Para evitar criar e destruir conexões TCP com o Redis a cada requisição HTTP, instanciamos um ConnectionPool assíncrono gerenciado pelo protocolo lifespan do FastAPI:

import json
import random
from typing import AsyncGenerator, Optional, Annotated
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException
import redis.asyncio as aioredis
from redis.asyncio import Redis, ConnectionPool

REDIS_URL = "redis://:senha_segura@redis-cluster.internal:6379/0"

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    # Cria pool de sockets assíncronos na inicialização
    pool = ConnectionPool.from_url(
        REDIS_URL,
        max_connections=50,
        decode_responses=True,      # Strings automáticas em vez de bytes
        socket_timeout=2.0,
        socket_connect_timeout=2.0,
    )
    app.state.redis_pool = pool
    app.state.redis_client = Redis(connection_pool=pool)
    
    yield
    
    # Encerramento gracioso do pool de conexões
    await app.state.redis_client.aclose()
    await pool.disconnect()

app = FastAPI(lifespan=lifespan)

# Dependência rápida injetável nos endpoints
async def get_redis_client() -> Redis:
    return app.state.redis_client

RedisDep = Annotated[Redis, Depends(get_redis_client)]

🛡️ 3. A Tríade de Patologias de Cache e Suas Mitigações

Ao projetar sistemas de alta escala, caches ingênuos podem colapsar a infraestrutura durante horários de pico. Existem três patologias clássicas:

1. Cache Stampede (Efeito Manada / Dogpile)

  • Problema: Um produto com alta visitação tem sua chave expirada no Redis. No exato instante da expiração, 5.000 requisições simultâneas percebem o Cache Miss e disparam a mesma query pesada contra o PostgreSQL ao mesmo tempo, causando travamento do banco por exaustão de conexões.
  • Mitigação Canônica (Lock Mutex com SET NX PX): Apenas uma requisição obtém a permissão de recalcular o valor; as demais aguardam uma fração de segundo e consomem o dado recém-gerado.

2. Cache Avalanche (Queda Simultânea em Cascata)

  • Problema: Todas as chaves são configuradas com TTL fixo de exatamente 300 segundos. Cinco minutos após uma carga inicial, todas as chaves expiram exatamente no mesmo segundo, descarregando 100% do tráfego sobre o banco.
  • Mitigação: Adicionar Jitter Aleatório ao tempo de expiração: 
    ttl_final = ttl_base + random.randint(-15, 15)  # Ex: entre 285s e 315s

3. Cache Penetration (Penetração Direta)

  • Problema: Um atacante consulta massivamente IDs inexistentes (/produtos/99999999). Como o dado não existe, nunca haverá Cache Hit, e todas as requisições chegam invariavelmente ao banco de dados relacional.
  • Mitigação: Armazenar o valor sentinela None ou string vazia com um TTL curto (ex: 60s) para bloquear consultas repetidas a entidades inexistentes.

💻 4. Implementação Completa do Padrão Cache-Aside Resiliente

async def obter_produto_cache_aside(
    produto_id: int,
    redis: Redis,
    session: AsyncSession
) -> dict:
    cache_key = f"cache:produto:{produto_id}"
    lock_key = f"lock:produto:{produto_id}"

    # 1. Tenta recuperar do Redis (Cache HIT)
    dado_em_cache = await redis.get(cache_key)
    if dado_em_cache is not None:
        if dado_em_cache == "__NOT_FOUND__":
            raise HTTPException(status_code=404, detail="Produto inexistente (cached)")
        return json.loads(dado_em_cache)

    # 2. Adquire lock distribuído exclusivo com timeout de 3 segundos
    adquiriu_lock = await redis.set(lock_key, "1", nx=True, px=3000)

    if not adquiriu_lock:
        # Outro processo já está calculando; aguarda 100ms e tenta novamente do cache
        import asyncio
        await asyncio.sleep(0.1)
        return await obter_produto_cache_aside(produto_id, redis, session)

    try:
        # 3. Consulta a fonte da verdade no banco relacional
        stmt = select(Produto).where(Produto.id == produto_id)
        resultado = await session.exec(stmt)
        produto = resultado.first()

        if not produto:
            # Proteção contra Cache Penetration: grava chave negativa com TTL curto (60s)
            await redis.set(cache_key, "__NOT_FOUND__", ex=60)
            raise HTTPException(status_code=404, detail="Produto inexistente")

        # 4. Grava no cache com TTL base de 5 minutos + Jitter
        payload_json = json.dumps(produto.model_dump())
        ttl_com_jitter = 300 + random.randint(0, 30)
        await redis.set(cache_key, payload_json, ex=ttl_com_jitter)

        return produto.model_dump()
    finally:
        # Libera o lock atômico
        await redis.delete(lock_key)