Como vimos neste artigo postado anteriormente aqui no blog, o Celery é uma ferramenta que permite executar tarefas de maneira assíncrona através de mensageria. Isso por si só já traz diversos benefícios, mas também é possível imaginarmos criando tarefas que seriam executadas dados certos intervalos de tempo (por exemplo o disparo de processos serem executados de madrugada) ou durante horários específicos. Este trabalho pode ser feito manualmente, basta criarmos um programa em Python que analisa todos os parâmetros temporais cadastrados e dispare as tarefas na hora adequada invocando a função através do método .delay().
Apesar de ser interessante desenvolver algo assim a ideia não é tão nova e o próprio pessoal que desenvolve o Celery se adiantou no assunto e criou o Celery Beat, uma aplicação que permite invocar tarefas do Celery conforme o tempo cadastrado.
Primeiro criamos e configuramos a nossa task, por fins de simplicidade iremos usar um Redis rodando localmente como broker (no post anterior usamos o SQS, porém como é mais complexo de configurar, o Redis já nos atende bem):
import os
import time
from celery import Celery
app = Celery('tasks', broker=os.getenv('BROKER_URL', 'redis://localhost/0'))
@app.task
def envia_email():
lista_de_usuarios = [
'Monica',
'Cebolinha',
'Magali',
'Cascão',
'Louco',
'Tina',
'Rolo',
]
for usuario in lista_de_usuarios:
print(f'Enviando email para {usuario}...')
time.sleep(1)
print(f'Email enviando para {usuario}!')
Após isso podemos fazer a invocação manualmente para testarmos. A saída será a invocação das funções:
Agora iremos para a parte mais importante deste artigo, como realizar o agendamento de tasks e a maneira mais simples é dentro do próprio arquivo de tasks definirmos a configuração. Para o nosso teste vamos chamar a task que criamos a cada 10 segundos (apenas para não termos que esperar muito tempo):
import os
import time
from celery import Celery
app = Celery('tasks', broker=os.getenv('BROKER_URL', 'redis://localhost/0'))
@app.task
def envia_email():
...
app.conf.beat_schedule = {
'envia-emails-a-cada-30-segundas': {
'task': 'tasks.envia_email',
'schedule': 10.0,
}
}
E executamos: celery -A tasks beat
Após um pequeno intervalo de tempo vemos a execução da nossa task como se tivéssemos chamados manualmente.
Para melhorar a nossa estrutura podemos separar a configuração do projeto, as tasks e o startup. Para isso basta separarmos alguns objetos em mais módulos e alterar algumas coisas simples:
## config.py
CELERY_BROKER_URL = 'redis://localhost/0'
CELERY_BEAT_SCHEDULE = {
'envia-emails-a-cada-30-segundas': {
'task': 'tasks.envia_email',
'schedule': 10.0,
}
}
## tasks.py
import time
from main import app
@app.task
def envia_email():
lista_de_usuarios = [
'Monica',
'Cebolinha',
'Magali',
'Cascão',
'Louco',
'Tina',
'Rolo',
]
for usuario in lista_de_usuarios:
print(f'Enviando email para {usuario}...')
time.sleep(1)
print(f'Email enviando para {usuario}!')
## main.py
from celery import Celery
app = Celery('tasks')
app.autodiscover_tasks(lambda: ['tasks'])
app.config_from_object('config', namespace='CELERY')
Com isso temos uma estrutura de tarefas agendadas que pode ir para produção. A estrutura de componentes pode ser vista abaixo, expandindo o exemplo que fizemos no post anterior:
