Tengo una aplicación simple de Django (v3.1) donde recibo datos de un formulario, los proceso con una vista y luego los paso a Celery (v4.4.7, RabbitMQ como intermediario). Dependiendo de los datos enviados en el formulario, puede ser una tarea única o una tarea periódica.
La tarea periódica debe ejecutar la misma tarea que la tarea de una sola vez, pero, bueno, con un horario periódico. Me gustaría pasar ese cronograma a la tarea, incluida una fecha de inicio, una fecha de finalización y un intervalo (por ejemplo: ejecutar cada 2 días a las 4 p. m., comenzando ahora hasta 4 semanas).
Mi vista (acortada y renombrada con fines ilustrativos, por supuesto):
# views.py if request.method == 'POST': form = BackupForm(request.POST) if form.is_valid(): data = ... if not form.cleaned_data['periodic']: # execute one-time task celery_task = single_task.delay(data) else: schedule = { 'first_backup': form.cleaned_data['first_backup'], 'last_backup': form.cleaned_data['last_backup'], 'intervall_every': form.cleaned_data['intervall_every'], 'intervall_unit': form.cleaned_data['intervall_unit'], 'intervall_time': form.cleaned_data['intervall_time'], } # execute periodic task, depending on the schedule submitted in the form celery_task = single_task.delay(data, schedule=schedule) return HttpResponseRedirect(reverse('app:index'))La tarea única se ve así:
# tasks.py @shared_task def single_task(data: dict, **kwargs) -> None: asyncio.run(bulk_screen(data=data)) # TODO: receive schedule if periodic and create a periodic task with itEsto funciona bien para la tarea individual. Sin embargo, no sé cómo adaptar esto para crear tareas periódicas dinámicas. Los datos de mi horario varían, dependiendo de la entrada del formulario de los usuarios. Tengo que crear la tarea periódica en tiempo de ejecución.
De acuerdo con la documentación oficial sobre tareas periódicas , crontab Schedules es lo que necesito:
from celery.schedules import crontab app.conf.beat_schedule = { # Executes every Monday morning at 7:30 am 'add-every-monday-morning': { 'task': 'tasks.add', 'schedule': crontab(hour=7, minute=30, day_of_week=1), 'args': (16, 16), }, }Aunque esto se ve bien, se encuentra en la configuración de apio con horarios codificados.
También leí sobre el decorador on_after_finalize.connect donde podría hacer algo como esto:
@celery_app.on_after_finalize.connect def setup_periodic_tasks(sender, **kwargs): sender.add_periodic_task(10.0, task123.s('hello'))Pero no sé cómo pasar el horario a esta función. Además, ¿cuál es el remitente? ¿Puedo pasarlo de mi vista?
Luego leí acerca de cómo completar los modelos relevantes en celery beat aquí . Pero supongo que tiene que haber una forma más elegante, usando la versión estable sin decoradores obsoletos.
Gracias.
Definitivamente deberías intentar llenar el PeriodicTask o CrontabSchedule proporcionado con el paquete django_celery_beat . enlace a documentos
Celery beat es el programador que se ejecuta periódicamente y simplemente ejecutará todas las tareas según un programa (una base de datos respaldada en el caso de django_celery_beat ). Referencia1 , Referencia2
Celery Beat es sin duda la forma más limpia de manejar tareas periódicas con diferentes horarios en lugar de crear su propio programador o manejar diferentes horarios.
Creo que en celery 4.x no puede configurar el cronograma dinámico que necesita (recuerdo haber visto hablar de que aparece en 5 pero no estoy seguro de si lo ha hecho). Sin embargo, su código parece prometedor: solo necesita verificar un cronograma en single_task y programar una nueva tarea (si es necesario) con una countdown o eta . La tarea puede programar su(s) propia(s) tarea(s) para un tiempo futuro .
p.ej
@shared_task def single_task(data: dict, **kwargs) -> None: asyncio.run(bulk_screen(data=data)) schedule = kwargs.get("schedule") if schedule: # do we need to run this again? next_run_at = get_next_run_at(schedule) if next_run_at: # yes! single_task.apply_async(args=[data], kwargs=kwargs, eta=next_run_at) # else the task will complete without rescheduling itself def next_run_at_schedule(schedule: dict) -> Optional[datetime]: """ return None if the schedule has expired, or is invalid else return the date and time to run the next task """ pass ConsulteCelery Docs para obtener detalles de apply_async y eta . No he probado esto, pero el principio es sólido. Debería
single_task para que las ejecuciones posteriores se programen en caso de un error dentro del código principal Si lo prefiere, puede usar la countdown en lugar de eta con apply_async.
Por supuesto, puede ser más fácil (y más confiable, ya que realizar correctamente el manejo de errores que mencioné no es trivial) cargar los horarios en una base de datos y luego ejecutar una tarea de apio simple cada minuto para verificar las copias de seguridad que deben realizarse: esto luego los dispararía en paralelo a través simple_task.delay(data) .