Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis

💻 Código de Exemplo 01 — Gravação e Leitura de Chaves com TTL usando redis.asyncio

# cache.py
from redis.asyncio import Redis, ConnectionPool

_pool = ConnectionPool.from_url("redis://localhost:6379/0", decode_responses=True, max_connections=20)

def get_redis() -> Redis:
    return Redis(connection_pool=_pool)

# main.py
import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis
from cache import get_redis

app = FastAPI()

@app.put("/perfis/{user_id}")
async def gravar_perfil_cache(user_id: int, redis: Redis = Depends(get_redis)) -> dict:
    dados = {"user_id": user_id, "nome": "Alice", "plano": "premium"}
    await redis.set(f"user:{user_id}:profile", json.dumps(dados), ex=60)
    return {"mensagem": "Perfil gravado no cache com TTL=60s"}

@app.get("/perfis/{user_id}")
async def ler_perfil_cache(user_id: int, redis: Redis = Depends(get_redis)) -> dict:
    bruto = await redis.get(f"user:{user_id}:profile")
    if bruto is None:
        return {"cache": "MISS"}
    return {"cache": "HIT", "dados": json.loads(bruto)}

💻 Código de Exemplo 02 — Cache-Aside com Invalidação Ativa em Update/Delete

import json
from fastapi import FastAPI, Depends, HTTPException
from redis.asyncio import Redis
from cache import get_redis

app = FastAPI()

# "Banco de dados" simulado em memória, para fins didáticos.
_banco_produtos: dict[int, dict] = {
    1: {"id": 1, "nome": "Teclado Mecânico", "preco": 349.90},
    2: {"id": 2, "nome": "Mouse Ergonômico", "preco": 189.50},
}

@app.get("/produtos/{produto_id}")
async def buscar_produto(produto_id: int, redis: Redis = Depends(get_redis)) -> dict:
    chave = f"produto:{produto_id}"
    cache_hit = await redis.get(chave)
    if cache_hit is not None:
        return {"origem": "cache", **json.loads(cache_hit)}

    produto = _banco_produtos.get(produto_id)
    if produto is None:
        raise HTTPException(status_code=404, detail="Produto não encontrado")

    await redis.set(chave, json.dumps(produto), ex=300)
    return {"origem": "banco", **produto}

@app.patch("/produtos/{produto_id}")
async def atualizar_produto(produto_id: int, novo_preco: float, redis: Redis = Depends(get_redis)) -> dict:
    if produto_id not in _banco_produtos:
        raise HTTPException(status_code=404, detail="Produto não encontrado")

    _banco_produtos[produto_id]["preco"] = novo_preco
    await redis.delete(f"produto:{produto_id}")  # invalidação ativa: purga a chave defasada
    return {"mensagem": "Produto atualizado; cache invalidado", "produto": _banco_produtos[produto_id]}

💻 Código de Exemplo 03 — Proteção contra Cache Stampede com Lock Distribuído

import asyncio
import json
from redis.asyncio import Redis

async def obter_com_protecao_stampede(redis: Redis, chave: str, ttl: int, factory) -> dict:
    """Evita que N requisições simultâneas recomputem a mesma chave quente ao expirar."""
    cached = await redis.get(chave)
    if cached is not None:
        return json.loads(cached)

    lock_key = f"lock:{chave}"
    # SET NX PX: só um processo consegue criar o lock; os demais aguardam e tentam de novo.
    obteve_lock = await redis.set(lock_key, "1", nx=True, px=5000)
    if not obteve_lock:
        await asyncio.sleep(0.1)
        return await obter_com_protecao_stampede(redis, chave, ttl, factory)

    try:
        # Revalida: outro processo pode ter terminado de gravar o cache enquanto esperávamos o lock.
        cached = await redis.get(chave)
        if cached is not None:
            return json.loads(cached)

        valor = await factory()
        await redis.set(chave, json.dumps(valor), ex=ttl)
        return valor
    finally:
        await redis.delete(lock_key)