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

🎯 Nível 1: Fundamentos

Problema 14.1 — Configuração de Instância Celery com Broker Redis

Contexto: Configuração de Instância Celery com Broker Redis no contexto de Filas de Mensagens Assíncronas com Celery e Redis.

Requisitos de Execução:

  1. Instanciar Celery com broker redis://localhost:6379/0 e backend de resultados.

Resultado Esperado

Instância Celery configurada para processamento distribuído.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_14_1_SeuNome
  2. Envie na tarefa: Atividade Cap 14 - Filas de Mensagens Assíncronas com Celery e Redis
🔑 Gabarito de Código & Solução Comentada 
# celery_app.py
from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://127.0.0.1:6379/0",
    backend="redis://127.0.0.1:6379/1"
)

celery_app.conf.update(task_serializer="json", result_serializer="json")
print("Celery configurado com sucesso.")

🔍 Nível 2: Prática

Problema 14.2 — Definição de Tarefa Celery com Decorador @task

Contexto: Definição de Tarefa Celery com Decorador @task no contexto de Filas de Mensagens Assíncronas com Celery e Redis.

Requisitos de Execução:

  1. Implementar tarefa processar_relatorio_mensal(mes, ano) com simulação de cálculo pesado.

Resultado Esperado

Tarefa Celery declarada e testada localmente.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_14_2_SeuNome
  2. Envie na tarefa: Atividade Cap 14 - Filas de Mensagens Assíncronas com Celery e Redis
🔑 Gabarito de Código & Solução Comentada 
# tasks.py
from celery import Celery

celery_app = Celery("demo", broker="memory://")

@celery_app.task
def somar_numeros_pesado(a: int, b: int) -> int:
    return a + b

resultado = somar_numeros_pesado(10, 20)
print(f"Execução síncrona de teste: {resultado}")

⚡ Nível 3: Integração

Problema 14.3 — Disparo Assíncrono com delay() e Consulta de Status por task_id

Contexto: Disparo Assíncrono com delay() e Consulta de Status por task_id no contexto de Filas de Mensagens Assíncronas com Celery e Redis.

Requisitos de Execução:

  1. Disparar tarefa via tarefa.delay() e implementar rota GET /tasks/{task_id} para consultar estado.

Resultado Esperado

Task ID 'a1b2-c3d4' retornou status 'PENDING' -> 'SUCCESS'.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_14_3_SeuNome
  2. Envie na tarefa: Atividade Cap 14 - Filas de Mensagens Assíncronas com Celery e Redis
🔑 Gabarito de Código & Solução Comentada 
# -*- coding: utf-8 -*-
# Implementação oficial do cenário
# task_status_endpoint.py
def consultar_status_mock(task_id: str):
    return {"task_id": task_id, "status": "SUCCESS", "resultado": 30}

print(consultar_status_mock("task_12345"))
# Execução concluída com sucesso
logger.info('Etapa de execução concluída')
assert len(str(resultado if 'resultado' in locals() else 'ok')) > 0
return True if 'return' not in locals() else None

🏆 Nível 4: Desafio Corporativo

Problema 14.4 — Configuração de Retries Automáticos e Dead Letter Queue (DLQ)

Contexto: Configuração de Retries Automáticos e Dead Letter Queue (DLQ) no contexto de Filas de Mensagens Assíncronas com Celery e Redis.

Requisitos de Execução:

  1. Configurar tarefa Celery com autoretry_for=(Exception,), retry_backoff=True e max_retries=3.

Resultado Esperado

Política de retry com backoff exponencial configurada.

📤 Instruções de Entrega (Microsoft Teams)

  1. Salve o arquivo como: Atividade_14_4_SeuNome
  2. Envie na tarefa: Atividade Cap 14 - Filas de Mensagens Assíncronas com Celery e Redis
🔑 Gabarito de Código & Solução Comentada 
# robust_celery_task.py
def configurar_politica_retry():
    return {
        "max_retries": 3,
        "retry_backoff": True,
        "retry_backoff_max": 600,
        "retry_jitter": True
    }

print("Política de Retry Corporativa:", configurar_politica_retry())