Spec Backend Com Python E Fastapi • Trilha Progressiva em 4 Níveis

🎯 Nível 1: Fundamentos

Problema 15.1 — Armazenamento e Recuperação de Chaves com TTL no Redis

Contexto: Armazenamento e Recuperação de Chaves com TTL no Redis no contexto de Caching Distribuído com Redis.

Requisitos de Execução:

  1. Criar rotina para conectar ao Redis, gravar chave com tempo de expiração (TTL de 60 segundos) e recuperar valor.

Resultado Esperado

Chave 'user:42:profile' gravada com TTL=60s e recuperada com sucesso.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_15_1_SeuNome
  2. Envie na tarefa: Atividade Cap 15 - Caching Distribuído com Redis
🔑 Gabarito de Código & Solução Comentada 
# redis_cache_basic.py
import json

class MiniRedisClient:
    def __init__(self): self.store = {}
    def set(self, k, v, ex=60): self.store[k] = (v, ex)
    def get(self, k): return self.store.get(k, (None, 0))[0]

r = MiniRedisClient()
r.set("cache_demo", json.dumps({"nome": "Alice"}), ex=60)
print("Cache Recuperado:", json.loads(r.get("cache_demo")))

🔍 Nível 2: Prática

Problema 15.2 — Padrão Cache-Aside (Lazy Loading)

Contexto: Padrão Cache-Aside (Lazy Loading) no contexto de Caching Distribuído com Redis.

Requisitos de Execução:

  1. Implementar função de busca que consulte primeiro o cache; em caso de Miss, consulte o banco e popule o cache.

Resultado Esperado

1ª Consulta: Cache MISS -> Banco consultado -> Cache populado.
2ª Consulta: Cache HIT imediato.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_15_2_SeuNome
  2. Envie na tarefa: Atividade Cap 15 - Caching Distribuído com Redis
🔑 Gabarito de Código & Solução Comentada 
# cache_aside.py
cache_mem = {}

def buscar_produto_com_cache(prod_id: int):
    chave = f"prod:{prod_id}"
    if chave in cache_mem:
        print("Cache HIT!")
        return cache_mem[chave]
    print("Cache MISS! Consultando banco de dados...")
    dado_banco = {"id": prod_id, "nome": f"Produto {prod_id}"}
    cache_mem[chave] = dado_banco
    return dado_banco

print(buscar_produto_com_cache(10))
print(buscar_produto_com_cache(10))

⚡ Nível 3: Integração

Problema 15.3 — Invalidação Ativa de Cache em Operações de Update/Delete

Contexto: Invalidação Ativa de Cache em Operações de Update/Delete no contexto de Caching Distribuído com Redis.

Requisitos de Execução:

  1. Garantir que ao atualizar ou deletar um produto, a respectiva chave de cache seja purgada (r.delete(key)).

Resultado Esperado

Produto atualizado no banco -> Chave de cache purgada com sucesso.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_15_3_SeuNome
  2. Envie na tarefa: Atividade Cap 15 - Caching Distribuído com Redis
🔑 Gabarito de Código & Solução Comentada 
# cache_invalidation.py
def atualizar_produto_e_invalidar(prod_id, novos_dados, cache_store):
    print(f"Atualizando produto {prod_id} no banco...")
    chave = f"prod:{prod_id}"
    if chave in cache_store:
        del cache_store[chave]
        print(f"Cache para chave '{chave}' invalidado.")

c_store = {"prod:5": {"id": 5, "nome": "Antigo"}}
atualizar_produto_e_invalidar(5, {"nome": "Novo"}, c_store)
print("Estado do Cache:", c_store)

🏆 Nível 4: Desafio Corporativo

Problema 15.4 — Prevenção de Problemas em Cache: Stampede, Avalanche e Penetration

Contexto: Prevenção de Problemas em Cache: Stampede, Avalanche e Penetration no contexto de Caching Distribuído com Redis.

Requisitos de Execução:

  1. Implementar técnica de Mutex Locking (Singleflight) para evitar que milhares de requisições sobrecarreguem o banco ao expirar uma chave quente.

Resultado Esperado

Cache Stampede prevenido: Apenas 1 thread consulta o banco enquanto as demais aguardam.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_15_4_SeuNome
  2. Envie na tarefa: Atividade Cap 15 - Caching Distribuído com Redis
🔑 Gabarito de Código & Solução Comentada 
# cache_stampede_guard.py
import threading

lock_recomputacao = threading.Lock()

def obter_dado_protegido(chave, cache_dict):
    if chave not in cache_dict:
        with lock_recomputacao:
            if chave not in cache_dict:  # Double-checked locking
                print("Thread única recomputando dado caro...")
                cache_dict[chave] = "DADO_RECOMPUTADO_VALIOSO"
    return cache_dict[chave]

cache_global = {}
print(obter_dado_protegido("chave_quente", cache_global))