Celery + Redis: tarefas assíncronas em Python

# requirements.txt
celery[redis]==5.4.0
redis==5.2.0

# app/celery_app.py
from celery import Celery
import os

celery_app = Celery(
    "tasks",
    broker=os.environ["REDIS_URL"],      # redis://redis:6379/0
    backend=os.environ["REDIS_URL"],     # armazena resultados
    include=["app.tasks"],               # módulos com tarefas
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="America/Sao_Paulo",
    enable_utc=True,
    task_track_started=True,
    result_expires=3600,                 # resultados expiram em 1h
    worker_prefetch_multiplier=1,        # processar 1 tarefa por vez
)

# app/tasks.py
from app.celery_app import celery_app
import time

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def enviar_email(self, destinatario: str, assunto: str, corpo: str):
    try:
        # lógica de envio de e-mail
        send_email(destinatario, assunto, corpo)
    except Exception as exc:
        raise self.retry(exc=exc)  # retry automático em caso de falha

services:
  api:
    build: .
    restart: always
    env_file: .env
    depends_on:
      redis:
        condition: service_healthy
      postgres:
        condition: service_healthy
    networks:
      - app_net

  celery_worker:
    build: .
    restart: always
    command: celery -A app.celery_app worker --loglevel=info --concurrency=4
    env_file: .env
    depends_on:
      redis:
        condition: service_healthy
    networks:
      - app_net

  celery_beat:
    build: .
    restart: always
    command: celery -A app.celery_app beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
    env_file: .env
    depends_on:
      - celery_worker
    networks:
      - app_net

  redis:
    image: redis:7-alpine
    restart: always
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      retries: 5
    networks:
      - app_net

networks:
  app_net:

# Chamadas de tarefas Celery

# 1. delay() — forma mais simples
from app.tasks import enviar_email
task = enviar_email.delay("user@email.com", "Bem-vindo!", "Corpo do e-mail")
print(task.id)  # UUID da tarefa para consultar resultado

# 2. apply_async() — mais controle
task = enviar_email.apply_async(
    args=["user@email.com", "Assunto"],
    kwargs={"corpo": "Corpo"},
    countdown=300,         # executar em 5 minutos
    eta=datetime(2026, 6, 8, 9, 0),  # executar na data/hora específica
    expires=3600,          # cancelar se não executar em 1h
    queue="emails",        # fila específica
)

# 3. Verificar status
from celery.result import AsyncResult
result = AsyncResult(task.id)
print(result.status)   # PENDING, STARTED, SUCCESS, FAILURE, RETRY
print(result.result)   # valor retornado pela tarefa

# 4. Em APIs FastAPI/Flask — não bloquear a resposta
@app.post("/usuarios")
async def criar_usuario(dados: UserCreate):
    user = await criar_user_db(dados)
    enviar_email.delay(user.email, "Bem-vindo!", gerar_corpo(user))
    return user  # retorna ANTES do e-mail ser enviado

# app/celery_app.py — configurar schedule
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    # Limpar sessões expiradas todo dia às 3h
    "limpar-sessoes": {
        "task": "app.tasks.limpar_sessoes_expiradas",
        "schedule": crontab(hour=3, minute=0),
    },
    # Enviar relatório semanal — segunda às 8h
    "relatorio-semanal": {
        "task": "app.tasks.enviar_relatorio",
        "schedule": crontab(hour=8, minute=0, day_of_week=1),
    },
    # Verificar saúde do sistema a cada 5 minutos
    "health-check": {
        "task": "app.tasks.verificar_saude",
        "schedule": 300.0,  # a cada 300 segundos
    },
}

# app/tasks.py — tarefas periódicas
@celery_app.task
def limpar_sessoes_expiradas():
    count = Session.objects.filter(expires_at__lt=datetime.now()).delete()
    logger.info(f"Limpas {count} sessões expiradas")

# Adicionar ao docker-compose.yml
  flower:
    build: .
    restart: always
    command: celery -A app.celery_app flower --port=5555
    env_file: .env
    ports:
      - "127.0.0.1:5555:5555"  # apenas localhost (acesse via SSH tunnel)
    depends_on:
      - redis
    networks:
      - app_net

# Acessar via SSH tunnel:
ssh -L 5555:localhost:5555 deploy@IP_DA_VPS
# Abrir: http://localhost:5555

# Monitorar via CLI (sem Flower)
celery -A app.celery_app inspect active   # tarefas em execução
celery -A app.celery_app inspect reserved # tarefas aguardando
celery -A app.celery_app inspect stats    # estatísticas dos workers