Tarefas Periodicas (Celery Beat)¶

Visao Geral¶

A aplicacao usa Celery Beat com DatabaseScheduler para gerenciar tarefas periodicas. Isso significa que as tarefas sao armazenadas no banco (django_celery_beat_periodictask) e podem ser adicionadas/removidas via admin sem reiniciar containers.

Tarefas Ativas¶

1. responses.tasks.send_scheduled_surveys¶

Logica: - Busca runs com is_scheduled=True, scheduled_at <= now(), status='pending', sent_at IS NULL - Para cada, envia o template via Meta Graph API - Atualiza sent_at = now() e is_scheduled = False - Logga sucesso/erro por run

2. responses.tasks.expire_stale_runs¶

Logica: Aplica 3 regras de expiracao (A, B, C) em massa.

3. celery.backend_cleanup (built-in do Celery)¶

Operacao¶

Listar tarefas ativas¶

1
2
3
4
5

docker compose exec backend python manage.py shell -c "
from django_celery_beat.models import PeriodicTask
for t in PeriodicTask.objects.all():
    print(f'  - {t.name} | task={t.task} | enabled={t.enabled} | interval={t.interval}')
"

Desabilitar uma tarefa temporariamente¶

1
2
3
4
5
6

docker compose exec backend python manage.py shell -c "
from django_celery_beat.models import PeriodicTask
t = PeriodicTask.objects.get(name='Expirar pesquisas antigas')
t.enabled = False
t.save()
"

(Nao precisa reiniciar o Celery Beat -- ele monitora o banco.)

Executar manualmente uma tarefa¶

1
2
3
4
5

docker compose exec backend python manage.py shell -c "
from responses.tasks import expire_stale_runs
result = expire_stale_runs.apply().get()  # apply() = sincrono
print(result)
"

Para executar via worker (assincrono):

1
2
3
4
5

docker compose exec backend python manage.py shell -c "
from responses.tasks import expire_stale_runs
async_result = expire_stale_runs.delay()
print(async_result.id)
"

Ver ultima execucao¶

1
2
3
4
5
6
7

docker compose exec backend python manage.py shell -c "
from django_celery_beat.models import PeriodicTask
for t in PeriodicTask.objects.all():
    print(f'{t.name}:')
    print(f'  last_run_at: {t.last_run_at}')
    print(f'  total_run_count: {t.total_run_count}')
"

Logs do Celery¶

Worker (executa as tarefas)¶

1
2

docker compose logs celery_worker -f
docker compose logs celery_worker --tail 50

Beat (scheduler que dispara as tarefas)¶

1

docker compose logs celery_beat -f

Restart do Celery¶

Em algumas situacoes (ex: trocou META_ACCESS_TOKEN no .env), pode ser necessario recriar os containers (nao apenas reiniciar):

1
2
3
4
5

# ❌ NAO faz pegar novo .env
docker compose restart celery_worker celery_beat

# ✅ Recarrega .env (recria container)
docker compose up -d --force-recreate celery_worker celery_beat

Adicionar Nova Tarefa Periodica¶

1. Implementar a task em <app>/tasks.py¶

1
2
3
4
5
6
7

@shared_task(bind=True, max_retries=3)
def minha_nova_task(self):
    try:
        # logica
        return {'status': 'success'}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

2. Criar migration para registrar no Celery Beat¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# <app>/migrations/000X_create_my_task.py
from django.db import migrations

def create_task(apps, schema_editor):
    PeriodicTask = apps.get_model('django_celery_beat', 'PeriodicTask')
    IntervalSchedule = apps.get_model('django_celery_beat', 'IntervalSchedule')

    schedule, _ = IntervalSchedule.objects.get_or_create(
        every=30,
        period='minutes',
    )

    PeriodicTask.objects.get_or_create(
        name='Minha nova task',
        defaults={
            'task': 'myapp.tasks.minha_nova_task',
            'interval': schedule,
            'enabled': True,
            'description': 'Descricao do que faz',
        }
    )

def reverse_task(apps, schema_editor):
    PeriodicTask = apps.get_model('django_celery_beat', 'PeriodicTask')
    PeriodicTask.objects.filter(name='Minha nova task').delete()

class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '000X-1_previous'),
        ('django_celery_beat', '__latest__'),
    ]

    operations = [
        migrations.RunPython(create_task, reverse_task),
    ]

3. Aplicar migration¶

1

docker compose exec backend python manage.py migrate

4. Reiniciar Beat para garantir que pegou a nova tarefa¶

1

docker compose restart celery_beat

Periodos Suportados (IntervalSchedule.period)¶

• microseconds
• seconds
• minutes
• hours
• days

Para horarios especificos (ex: "todo dia as 03:00 UTC"), usar CrontabSchedule em vez de IntervalSchedule.

Boas Praticas¶

1. Idempotencia: tarefas devem ser seguras para reexecutar (use filtros que excluem itens ja processados).
2. Timeout: respeitar CELERY_TASK_TIME_LIMIT=600 (10min) configurado.
3. Retry com backoff: usar self.retry(exc=exc, countdown=60) em caso de erro temporario.
4. Logging: usar logger.info() para sucesso, logger.error() para falhas. Evitar print().
5. Bulk operations: usar .update() em vez de iterar com .save() sempre que possivel.
6. select_related() / prefetch_related() quando a task itera sobre QuerySets com FK.

Troubleshooting¶

Tarefa nao esta executando¶

Possiveis causas: 1. enabled=False na PeriodicTask 2. Celery Beat parado (docker compose ps celery_beat) 3. Worker parado (Beat dispara mas worker nao consome) 4. Redis sem conexao (verificar redis://redis:6379/2 no settings) 5. Bug na task fazendo retry infinito (verificar logs)

Tarefa executando mais que o esperado¶

Verificar total_run_count e last_run_at. Se intervalo muito curto, pode ser: 1. interval.every mal configurado 2. Dois containers de Beat rodando (NUNCA fazer isso)

Migration criando tarefa duplicada¶

Sempre use get_or_create (nao create) e teste rollback com migrate <app> 000X-1.