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["Cliente HTTP / Webhook"] --> B["FastAPI API Gateway (Produtor)"]
    B -->|"task.delay(params) [Sem Bloqueio]"| C[("Broker: Redis DB 0 / RabbitMQ")]
    B -->|"HTTP 202 Accepted + task_id"| A
    
    C -->|"Pull / Consumo Concorrente"| D["Celery Worker 1 (Processo Isolado)"]
    C -->|"Pull / Consumo Concorrente"| E["Celery Worker 2 (Processo Isolado)"]
    
    D --> F{"Processamento da Tarefa"}
    F -->|"Sucesso"| G[("Result Backend: Redis DB 1")]
    F -->|"Falha Transitória"| H["Retry com Backoff Exponencial"]
    H --> C
    F -->|"Esgotou Retries"| I["Dead-Letter Queue / Alerta"]
    
    G --> J["Polling do Cliente: GET /tasks/{task_id}"]

    subgraph SCHEDULER["Agendamento Recorrente (Cron)"]
        K["Celery Beat"] -->|"Enfileira nos horários programados"| C
    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:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style E fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style F fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style G fill:#e0f2f1,stroke:#009688,stroke-width:2px
    style H fill:#fffde7,stroke:#fbc02d,stroke-width:2px
    style I fill:#ffebee,stroke:#e53935,stroke-width:2px
    style J fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style K fill:#fff3e0,stroke:#ff9800,stroke-width:2px

🏛️ 1. A Tríade do Processamento Distribuído: Produtor, Broker e Worker

Quando uma operação de backend consome alto uso de CPU (como processamento de imagens, compilação de relatórios analíticos em PDF, ingestão de dados massivos ou treinamento de modelos de Machine Learning), executá-la no processo principal do Uvicorn congela o Event Loop, impedindo o atendimento a outros clientes.

Para escalar horizontalmente, adotamos a arquitetura de Filas Distribuídas com Celery:

  1. Produtor (FastAPI): Recebe a requisição HTTP do usuário, valida o payload, publica uma mensagem de comando no Broker em menos de 5ms e devolve imediatamente um HTTP 202 Accepted acompanhado de um identificador único de tarefa (task_id).
  2. Broker de Mensagens (Redis / RabbitMQ): Atua como o buffer intermediário de alta confiabilidade. Garante persistência das mensagens em memória/disco mesmo se os servidores reiniciarem.
  3. Workers Celery: Processos independentes do Python (que podem residir em servidores dedicados com CPUs especializadas) que consomem as mensagens da fila, executam as funções pesadas e gravam o resultado no Result Backend.

Comparativo: Escolha de Broker e Result Backend

Tecnologia Papel Ideal Latência Média Persistência / Resiliência Quando Escolher
Redis Broker + Result Backend Ultra-baixa (< 1ms) RDB / AOF configurável Excelente para a imensa maioria dos casos (praticidade e velocidade).
RabbitMQ Broker Dedicado Baixa (~2ms) Muito alta (Erlang AMQP 0-9-1) Filas corporativas complexas com roteamento avançado (Exchange tópicos).
PostgreSQL Result Backend Média (~10ms) Confiabilidade ACID relacional Quando não se deseja adicionar Redis à infraestrutura existente.

⚙️ 2. Configuração e Implementação Idiomática

# worker/celery_app.py
from celery import Celery

celery_app = Celery(
    "tarefas_empresa",
    broker="redis://localhost:6379/0",     # Banco lógico 0 para a fila
    backend="redis://localhost:6379/1"    # Banco lógico 1 para os resultados
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="America/Sao_Paulo",
    enable_utc=True,
    task_track_started=True,               # Marca status 'STARTED' para rastreamento
    task_time_limit=300,                  # Hard timeout: mata a tarefa se passar de 5 min
    task_soft_time_limit=240,             # Soft timeout: levanta exceção para fechamento limpo
)

Definição de Tarefa com Retries e Backoff Exponencial

# worker/tasks.py
import time
from celery.utils.log import get_task_logger
from worker.celery_app import celery_app

logger = get_task_logger(__name__)

@celery_app.task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,          # Esperas: 1s, 2s, 4s, 8s...
    retry_jitter=True,           # Adiciona ruído aleatório para evitar colisão manada
    max_retries=5,
    name="tarefas.processar_faturamento"
)
def processar_faturamento_em_lote(self, lote_id: str) -> dict:
    logger.info(f"Iniciando faturamento do lote: {lote_id} (Tentativa {self.request.retries})")
    
    # Simulação de processamento pesado de CPU
    time.sleep(10)
    
    return {
        "lote_id": lote_id,
        "registros_faturados": 1250,
        "status": "CONCLUIDO"
    }

🚀 3. Roteamento no FastAPI e Consulta de Estado (AsyncResult)

No FastAPI, injetamos a chamada via .delay() (ou .apply_async() quando é necessário configurar tempo de expiração ou fila específica):

from fastapi import FastAPI, HTTPException, status
from celery.result import AsyncResult
from pydantic import BaseModel
from worker.tasks import processar_faturamento_em_lote
from worker.celery_app import celery_app

app = FastAPI()

class FaturamentoResponse(BaseModel):
    task_id: str
    status: str
    mensagem: str

@app.post("/faturamentos/{lote_id}", response_model=FaturamentoResponse, status_code=status.HTTP_202_ACCEPTED)
async def despachar_faturamento(lote_id: str):
    # Publica no broker e libera a thread do FastAPI imediatamente
    tarefa = processar_faturamento_em_lote.delay(lote_id)
    return FaturamentoResponse(
        task_id=tarefa.id,
        status=tarefa.status,
        mensagem="Processamento enfileirado com sucesso. Monitore via /faturamentos/status/{task_id}"
    )

@app.get("/faturamentos/status/{task_id}")
async def obter_status_tarefa(task_id: str):
    resultado = AsyncResult(task_id, app=celery_app)
    
    resposta = {
        "task_id": task_id,
        "status": resultado.status,  # PENDING, STARTED, RETRY, SUCCESS, FAILURE
    }
    
    if resultado.successful():
        resposta["resultado"] = resultado.result
    elif resultado.failed():
        resposta["erro"] = str(resultado.result)
        
    return resposta

⏰ 4. Agendamento com Celery Beat e Monitoramento com Flower

Para rotinas periódicas (como fechamento de balanço diário à meia-noite), o componente Celery Beat funciona como um agendador distribuído de tarefas:

# Configuração do Celery Beat no celery_app.py
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "expurgar-tokens-expirados-meia-noite": {
        "task": "tarefas.limpar_tokens",
        "schedule": crontab(hour=0, minute=0),
    },
    "sincronizar-estoque-a-cada-15-min": {
        "task": "tarefas.sincronizar_estoque",
        "schedule": 900.0,  # a cada 15 minutos em segundos
    },
}

Para observabilidade em produção, a ferramenta Flower (celery -A worker.celery_app flower --port=5555) disponibiliza um painel gráfico web em tempo real com métricas completas de tempo de execução, saturação de filas, taxa de falhas e capacidade de revogar tarefas travadas.