Celery + Redis: tarefas assíncronas em Python
Celery é a biblioteca padrão para processamento assíncrono em Python — permite executar tarefas pesadas (envio de e-mail, processamento de imagem, chamadas a APIs externas) em background sem bloquear a requisição HTTP. Com Redis como broker e Docker Compose, o setup completo leva menos de 30 minutos.
Instalar e configurar Celery
Celery usa Redis como fila de mensagens (broker). Configure a conexão e defina as tarefas:
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 falhaDica
bind=True passa a instância da tarefa como primeiro argumento (self), necessário para acessar self.retry(). max_retries=3 e default_retry_delay=60 fazem a tarefa tentar novamente 3 vezes com 60 segundos de intervalo em caso de falha.
docker-compose.yml com Celery worker
Cada worker Celery é um container separado — escale horizontalmente adicionando mais workers:
yaml
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:Chamar tarefas assíncronas
Tarefas Celery são chamadas de forma assíncrona — a requisição retorna imediatamente e o worker processa em background:
python
# 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 enviadoCelery Beat: tarefas periódicas
Celery Beat é o agendador — substitui cron para tarefas que precisam do contexto Python:
python
# 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")Monitorar filas com Flower
Flower é o painel web para monitorar workers e tarefas Celery:
bash
# 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$ runstack deploy --plan starter
Não quer configurar manualmente?
Não quer configurar manualmente? Implante o VPS em menos de 3 minutos com a Runstack. Infraestrutura da OPEN DATACENTER, com servidores no Brasil.