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

💻 Código de Exemplo 01 — Configuração da Instância Celery com Broker e Backend Redis

# celery_app.py
from celery import Celery

celery_app = Celery(
    "loja_worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
    include=["tasks"],
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=3600,       # resultados somem do Redis após 1h
    task_track_started=True,   # expõe o status STARTED, não só PENDING/SUCCESS
    worker_prefetch_multiplier=1,  # evita que 1 worker reserve várias tarefas pesadas de uma vez
)

💻 Código de Exemplo 02 — Tarefa com Retry Automático e Endpoint de Consulta de Status

# tasks.py
import time
from celery_app import celery_app

@celery_app.task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
)
def processar_pedido(self, pedido_id: int) -> dict:
    """Simula a baixa de estoque e emissão de nota fiscal de um pedido."""
    time.sleep(2)
    if pedido_id % 7 == 0:
        raise ConnectionError("Falha simulada ao contatar serviço fiscal externo")
    return {"pedido_id": pedido_id, "nota_fiscal": f"NF-{pedido_id:06d}", "status": "emitida"}

# main.py
from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from celery_app import celery_app
from tasks import processar_pedido

app = FastAPI(title="API de Pedidos")

@app.post("/pedidos/{pedido_id}/processar", status_code=202)
def disparar_processamento(pedido_id: int) -> dict:
    task = processar_pedido.delay(pedido_id)
    return {"task_id": task.id, "status": "PENDING"}

@app.get("/tasks/{task_id}")
def status_tarefa(task_id: str) -> dict:
    result = AsyncResult(task_id, app=celery_app)
    if result.status == "FAILURE":
        raise HTTPException(status_code=500, detail=str(result.result))
    return {
        "task_id": task_id,
        "status": result.status,
        "resultado": result.result if result.successful() else None,
    }

💻 Código de Exemplo 03 — Tarefas Periódicas com Celery Beat

# celery_app.py (trecho adicional de agendamento)
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "relatorio-diario-de-vendas": {
        "task": "tasks.gerar_relatorio_vendas",
        "schedule": crontab(hour=2, minute=0),  # todos os dias às 02h00
    },
    "limpeza-carrinho-abandonado": {
        "task": "tasks.limpar_carrinhos_abandonados",
        "schedule": crontab(minute=0, hour="*/6"),  # a cada 6 horas
    },
}

# Terminais separados necessários em desenvolvimento:
celery -A celery_app worker --loglevel=info   # processa as tarefas da fila
celery -A celery_app beat --loglevel=info     # apenas enfileira nos horários configurados