Celery: Filas de Tarefas e Processamento Assíncrono

Algumas operações são pesadas demais para acontecer durante uma requisição HTTP — enviar e-mails em massa, gerar relatórios PDF, processar imagens, treinar modelos de machine learning, sincronizar dados com APIs externas. Fazer o usuário esperar por tudo isso é inaceitável. Celery resolve esse problema: ele executa tarefas em segundo plano, de forma distribuída, com suporte a agendamento, retentativas automáticas e monitoramento. É o padrão da indústria para processamento assíncrono em Python.

Arquitetura do Celery

┌─────────────┐     enfileira     ┌─────────────┐     consome     ┌─────────────┐
│   Aplicação │ ────────────────► │   Broker    │ ──────────────► │   Worker    │
│  (FastAPI/  │                   │ (Redis/     │                 │  (Celery)   │
│   Flask)    │                   │  RabbitMQ)  │                 │             │
└─────────────┘                   └─────────────┘                 └──────┬──────┘
       ▲                                                                  │
       │                          ┌─────────────┐                        │ armazena
       │     lê resultado         │   Backend   │ ◄──────────────────────┘ resultado
       └────────────────────────  │   (Redis)   │
                                  └─────────────┘

• Producer — a aplicação que enfileira tarefas
• Broker — fila de mensagens (Redis ou RabbitMQ)
• Worker — processo que consome e executa as tarefas
• Backend — armazena resultados das tarefas (Redis, banco de dados)

Instalação

pip install celery redis
pip install flower   # monitoramento web

# Redis via Docker
docker run -d --name redis -p 6379:6379 redis:7-alpine

Configuração Básica

# celery_app.py
import os
from celery import Celery

# No Docker o Redis não está em localhost: a URL vem do ambiente
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")

def criar_celery(broker_url=f"{REDIS_URL}/0",
                 backend_url=f"{REDIS_URL}/1"):
    app = Celery(
        "escola",
        broker=broker_url,
        backend=backend_url,
        # módulo fora desta lista não registra tarefa no worker
        include=["tarefas.email", "tarefas.relatorio",
                 "tarefas.dados", "tarefas.agendadas"]
    )

    app.conf.update(
        # Serialização
        task_serializer         = "json",
        result_serializer       = "json",
        accept_content          = ["json"],

        # Timezone
        timezone                = "America/Sao_Paulo",
        enable_utc              = True,

        # Comportamento das tarefas
        task_track_started      = True,
        task_acks_late          = True,      # confirma após execução
        worker_prefetch_multiplier = 1,      # uma tarefa por vez por worker

        # Retentativas — "task_max_retries" não existe como configuração:
        # o Celery aceita o nome errado e o ignora. O padrão global vai por anotação.
        task_annotations        = {"*": {"max_retries": 3,
                                         "default_retry_delay": 60}},

        # Resultados
        result_expires          = 3600,      # 1 hora

        # Filas
        task_default_queue      = "padrao",
        task_queues             = {
            "padrao":    {"exchange": "padrao"},
            "prioritaria": {"exchange": "prioritaria"},
            "pesada":    {"exchange": "pesada"},
        }
    )

    return app


celery = criar_celery()

Definindo Tarefas

# tarefas/email.py
from celery_app import celery
from celery.utils.log import get_task_logger
import time

logger = get_task_logger(__name__)


@celery.task(
    name="email.enviar_boas_vindas",
    bind=True,          # acesso a self (instância da tarefa)
    max_retries=3,
    default_retry_delay=30,
    queue="prioritaria"
)
def enviar_boas_vindas(self, email: str, nome: str):
    """Envia e-mail de boas-vindas para novo usuário."""
    logger.info(f"Enviando boas-vindas para {email}")

    try:
        # Simulando envio de e-mail
        time.sleep(2)

        if "@invalido" in email:
            raise ValueError(f"Domínio inválido: {email}")

        logger.info(f"E-mail enviado com sucesso para {email}")
        return {"status": "enviado", "destinatario": email}

    except Exception as exc:
        logger.error(f"Falha ao enviar para {email}: {exc}")
        raise self.retry(exc=exc, countdown=30 * (self.request.retries + 1))


@celery.task(name="email.enviar_em_massa", queue="pesada")
def enviar_em_massa(destinatarios: list, assunto: str, corpo: str):
    """Envia e-mail para múltiplos destinatários."""
    enviados  = 0
    falhas    = []

    for dest in destinatarios:
        try:
            time.sleep(0.1)   # simulando envio
            enviados += 1
            logger.info(f"Enviado para {dest}")
        except Exception as e:
            falhas.append({"email": dest, "erro": str(e)})

    return {
        "total":    len(destinatarios),
        "enviados": enviados,
        "falhas":   len(falhas),
        "detalhes": falhas
    }

# tarefas/relatorio.py
from celery_app import celery
from celery.utils.log import get_task_logger
import time

logger = get_task_logger(__name__)


@celery.task(
    name="relatorio.gerar_pdf",
    bind=True,
    queue="pesada",
    time_limit=300       # mata a tarefa após 5 minutos
)
def gerar_relatorio_pdf(self, aluno_ids: list, periodo: str):
    """Gera relatório PDF de alunos — tarefa longa."""
    total    = len(aluno_ids)
    logger.info(f"Gerando relatório para {total} alunos")

    for i, aluno_id in enumerate(aluno_ids, 1):
        # Atualiza progresso
        self.update_state(
            state="PROGRESS",
            meta={
                "atual":     i,
                "total":     total,
                "percentual": round(i / total * 100, 1),
                "aluno_id":  aluno_id
            }
        )
        time.sleep(0.5)   # simula processamento

    caminho = f"/relatorios/relatorio_{periodo}.pdf"
    logger.info(f"Relatório gerado: {caminho}")

    return {
        "status":    "concluido",
        "caminho":   caminho,
        "total":     total,
        "periodo":   periodo
    }

Integrando com FastAPI

# main.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, EmailStr
from celery.result import AsyncResult
from celery_app import celery
from tarefas.email import enviar_boas_vindas, enviar_em_massa
from tarefas.relatorio import gerar_relatorio_pdf
from typing import List

app = FastAPI(title="API com Celery")


class NovoUsuario(BaseModel):
    nome:  str
    email: EmailStr


class EnvioMassa(BaseModel):
    destinatarios: List[EmailStr]
    assunto:       str
    corpo:         str


class SolicitacaoRelatorio(BaseModel):
    aluno_ids: List[int]
    periodo:   str


# Enfileirar tarefa e retornar ID
@app.post("/usuarios/registro", status_code=201)
async def registrar_usuario(dados: NovoUsuario):
    # Tarefa assíncrona — não bloqueia a resposta
    tarefa = enviar_boas_vindas.delay(dados.email, dados.nome)

    return {
        "mensagem":  "Usuário registrado. E-mail de boas-vindas será enviado.",
        "tarefa_id": tarefa.id
    }


@app.post("/emails/massa")
async def envio_em_massa(dados: EnvioMassa):
    tarefa = enviar_em_massa.apply_async(
        args=[dados.destinatarios, dados.assunto, dados.corpo],
        queue="pesada",
        countdown=5          # aguarda 5 segundos antes de iniciar
    )
    return {
        "mensagem":     f"Envio em massa para {len(dados.destinatarios)} destinatários enfileirado.",
        "tarefa_id":    tarefa.id,
        "total":        len(dados.destinatarios)
    }


@app.post("/relatorios/gerar")
async def solicitar_relatorio(dados: SolicitacaoRelatorio):
    tarefa = gerar_relatorio_pdf.apply_async(
        args=[dados.aluno_ids, dados.periodo],
        queue="pesada"
    )
    return {
        "mensagem":  "Geração de relatório iniciada.",
        "tarefa_id": tarefa.id
    }


# Verificar status de qualquer tarefa
@app.get("/tarefas/{tarefa_id}")
async def verificar_tarefa(tarefa_id: str):
    resultado = AsyncResult(tarefa_id, app=celery)

    resposta = {
        "id":     tarefa_id,
        "status": resultado.status,
    }

    if resultado.status == "PROGRESS":
        resposta["progresso"] = resultado.info

    elif resultado.status == "SUCCESS":
        resposta["resultado"] = resultado.get()

    elif resultado.status == "FAILURE":
        resposta["erro"] = str(resultado.result)

    # Id inventado ou resultado já expirado (result_expires) também
    # responde PENDING: o backend não distingue "na fila" de "nunca existiu"
    return resposta


# Revogar (cancelar) uma tarefa
@app.delete("/tarefas/{tarefa_id}")
async def cancelar_tarefa(tarefa_id: str):
    celery.control.revoke(tarefa_id, terminate=True)
    return {"mensagem": f"Tarefa {tarefa_id} cancelada."}

Agendamento de Tarefas: Celery Beat

# celery_app.py — adicionando agendamento, depois de celery = criar_celery()
from celery.schedules import crontab

celery.conf.beat_schedule = {
    # A cada hora
    "backup-banco-horario": {
        "task":     "dados.backup_banco",
        "schedule": 3600,
        "args":     ("escola",)
    },

    # Todo dia às 6h
    "relatorio-diario": {
        "task":     "relatorio.gerar_diario",
        "schedule": crontab(hour=6, minute=0),
    },

    # Toda segunda às 8h30
    "resumo-semanal": {
        "task":     "email.enviar_resumo_semanal",
        "schedule": crontab(hour=8, minute=30, day_of_week="mon"),
    },

    # Primeiro dia do mês
    "relatorio-mensal": {
        "task":     "relatorio.gerar_mensal",
        "schedule": crontab(day_of_month=1, hour=7, minute=0),
    },

    # A cada 30 minutos
    "sincronizar-dados": {
        "task":     "dados.sincronizar",
        "schedule": crontab(minute="*/30"),
    },
}

# tarefas/agendadas.py
from celery_app import celery
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@celery.task(name="dados.backup_banco")
def backup_banco(nome_banco: str):
    logger.info(f"Iniciando backup de {nome_banco}")
    # lógica de backup...
    return {"status": "ok", "banco": nome_banco}


@celery.task(name="relatorio.gerar_diario")
def gerar_relatorio_diario():
    from datetime import date
    hoje = date.today().isoformat()
    logger.info(f"Gerando relatório diário: {hoje}")
    # lógica do relatório...
    return {"data": hoje, "status": "gerado"}


@celery.task(name="email.enviar_resumo_semanal")
def enviar_resumo_semanal():
    logger.info("Enviando resumo semanal para gestores")
    # busca dados, monta e-mail, envia...
    return {"destinatarios": 15, "status": "enviado"}


@celery.task(name="relatorio.gerar_mensal")
def gerar_relatorio_mensal():
    logger.info("Gerando relatório mensal")
    return {"status": "gerado"}


@celery.task(name="dados.sincronizar")
def sincronizar():
    logger.info("Sincronizando com o sistema externo")
    return {"status": "ok"}

Chains, Groups e Chords: Workflows Complexos

from celery import chain, group, chord
from tarefas.dados import buscar_dados, processar_chunk, consolidar


# CHAIN — tarefas em sequência (saída de uma é entrada da próxima)
pipeline = chain(
    buscar_dados.s(periodo="2024-03"),
    processar_chunk.s(),
    consolidar.s()
)
resultado = pipeline.delay()


# GROUP — tarefas em paralelo
processamentos = group(
    processar_chunk.s(chunk) for chunk in range(1, 6)
)
resultados = processamentos.delay()


# CHORD — paralelo com callback ao final
workflow = chord(
    group(processar_chunk.s(i) for i in range(1, 6)),
    consolidar.s()   # chamado quando todos terminam
)
workflow.delay()


# Exemplo prático — pipeline de importação
from tarefas.importacao import (
    validar_csv,
    transformar_dados,
    inserir_banco,
    notificar_conclusao
)

def importar_arquivo(caminho_csv: str, usuario_email: str):
    pipeline = chain(
        validar_csv.s(caminho_csv),
        transformar_dados.s(),
        inserir_banco.s(),
        notificar_conclusao.s(usuario_email)
    )
    return pipeline.apply_async()

Monitoramento com Flower

# Iniciando o Flower
celery -A celery_app flower --port=5555

# Acesse: http://localhost:5555

# docker-compose.yml — adicionando Celery e Flower
services:
  redis:
    image: redis:7-alpine

  worker-padrao:
    build: .
    command: celery -A celery_app worker -Q padrao -c 4 --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  worker-pesado:
    build: .
    command: celery -A celery_app worker -Q pesada -c 2 --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  worker-prioritario:
    build: .
    command: celery -A celery_app worker -Q prioritaria -c 8 --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  beat:
    build: .
    command: celery -A celery_app beat --loglevel=info   # uma instância só
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis

  flower:
    build: .
    command: celery -A celery_app flower --port=5555
    environment:
      - REDIS_URL=redis://redis:6379
    ports:
      - "5555:5555"
    depends_on:
      - redis

Boas Práticas

# 1. Tarefas idempotentes — podem ser executadas múltiplas vezes sem efeito colateral
@celery.task(name="email.confirmar_cadastro")
def confirmar_cadastro(usuario_id: int):
    from models import Usuario
    usuario = Usuario.buscar(usuario_id)

    if usuario.email_confirmado:
        return {"status": "ignorado", "motivo": "já confirmado"}

    # só processa se ainda não confirmado
    usuario.confirmar_email()
    return {"status": "confirmado"}


# 2. Sempre defina timeout
@celery.task(name="dados.processar", time_limit=120, soft_time_limit=100)
def processar_dados(dados_id: int):
    from celery.exceptions import SoftTimeLimitExceeded
    try:
        # processamento...
        pass
    except SoftTimeLimitExceeded:
        # cleanup antes do hard limit
        logger.warning(f"Soft limit atingido para dados {dados_id}")
        raise


# 3. Não passe objetos complexos — passe IDs
# ERRADO
@celery.task
def processar_usuario_errado(usuario):   # objeto inteiro
    pass

# CORRETO
@celery.task
def processar_usuario_correto(usuario_id: int):   # apenas o ID
    usuario = buscar_usuario_do_banco(usuario_id)
    pass


# 4. Logging estruturado
@celery.task(bind=True, name="dados.importar")
def importar_dados(self, arquivo_id: int):
    logger.info("Iniciando importação",
                extra={"tarefa_id": self.request.id, "arquivo_id": arquivo_id})

Celery tira da requisição o trabalho que não cabe nela: a aplicação publica uma mensagem no broker e responde na hora, um worker em outro processo executa a tarefa e grava o resultado no backend, e o cliente consulta o andamento pelo id. Filas separadas deixam o e-mail de boas-vindas passar na frente do relatório pesado, o Beat substitui o cron com agendamentos declarados no próprio código, e chain, group e chord montam fluxos em que uma etapa alimenta a outra. Para o que é curto e pode se perder num reinício, as BackgroundTasks de FastAPI: APIs modernas com tipagem e documentação automática bastam; o Celery entra quando a tarefa precisa sobreviver ao processo que a pediu.

Os defeitos mais caros aqui não geram erro. Uma configuração com nome errado é aceita e ignorada, uma tarefa fora do include só falha no worker, um id que não existe responde PENDING para sempre, e um segundo Beat dobra cada envio. Por isso as tarefas precisam ser idempotentes, receber ids em vez de objetos e ter limite de tempo, e o Beat precisa rodar em um único processo. O Flower mostra o que está acontecendo, mas não sabe o que deveria ter acontecido.

Fontes e leituras recomendadas

• Celery — documentação oficial — https://docs.celeryq.dev/en/stable/
• Celery com FastAPI — guia prático — https://testdriven.io/blog/fastapi-and-celery/
• Flower — monitoramento — https://flower.readthedocs.io/en/latest/
• Redis como broker — https://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/redis.html
• Canvas — workflows com chain, group, chord — https://docs.celeryq.dev/en/stable/userguide/canvas.html
• PERCIVAL, Harry; GREGORY, Bob. Architecture Patterns with Python. O'Reilly Media, 2020. Cap. 11 — event-driven architecture e processamento assíncrono.
• GRIGORIK, Ilya. High Performance Browser Networking. O'Reilly Media, 2013. — contexto de performance para entender quando usar processamento assíncrono

Exercícios

Exercício 1

A equipe configurou o Celery com task_max_retries = 3 e task_default_retry_delay = 60 no app.conf.update, e as tarefas chamam self.retry(exc=exc) sem countdown. Em produção, o log mostra três minutos entre uma tentativa e outra, e ninguém encontra de onde sai esse número. O worker sobe sem nenhum aviso. O que aconteceu, e como fixar o padrão para todas as tarefas?

Exercício 2

O backup noturno agendado no Beat não roda há uma semana. O log do Beat mostra Sending due task backup-banco-horario no horário certo, e a tarefa está escrita e testada em tarefas/agendadas.py. No log do worker aparece Received unregistered task of type 'dados.backup_banco'. O desenvolvedor jura que o nome está idêntico nos dois lugares. Onde está o problema?

Exercício 3

Um cliente abre chamado: o relatório que ele pediu ontem "está na fila até agora". O suporte consulta /tarefas/{id} e vê "status": "PENDING". Os workers estão saudáveis, a fila está vazia e o Flower mostra que a tarefa com esse id terminou com sucesso ontem, em quarenta segundos. Explique a contradição e o que mudar na API.

Exercício 4

Depois de uma migração para Kubernetes, o e-mail de resumo semanal passou a chegar em dobro para os gestores, e a sincronização com o ERP roda duas vezes a cada meia hora. Nenhum código mudou. O deployment que antes tinha um contêiner com celery worker e outro com celery beat agora tem duas réplicas de cada um, "por alta disponibilidade". Qual réplica causou a duplicidade, e como resolver sem perder a redundância dos workers?

Exercício 5

Uma rota de FastAPI passa a dar erro 500 depois de uma refatoração que trocou processar.delay(usuario.id) por processar.delay(usuario), "para economizar uma consulta". Outra rota, que envia {"ids": {3, 7}}, também quebra. Numa terceira, a tarefa recebe {1: "a"} e depois faz dados[1], com KeyError no worker. Explique os três casos e o que efetivamente pode ser passado.