python - beat_schedule - Cómo agregar/eliminar dinámicamente tareas periódicas a apio(apio de cebo)

django beat_schedule

Question

Si tengo una función definida de la siguiente manera:

def add(x,y):
return x+y

¿Hay alguna manera de agregar dinámicamente esta función como una PeriodicTask de apio y ponerla en marcha en tiempo de ejecución? Me gustaría poder hacer algo como (pseudocódigo):

some_unique_task_id = celery.beat.schedule_task(add, run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)

También me gustaría detener o eliminar esa tarea de forma dinámica con algo así como (pseudocódigo):

celery.beat.remove_task(some_unique_task_id)

o

celery.beat.stop(some_unique_task_id)

FYI No estoy usando djcelery, que te permite administrar tareas periódicas a través del administrador de django.

Answer 1

Esta pregunta fue respondida en grupos de google .

NO SOY EL AUTOR, todo el mérito es para Jean Mark

Aquí hay una solución adecuada para esto. Trabajo confirmado. En mi caso, clasifiqué la tarea periódica y creé un modelo, ya que puedo agregar otros campos al modelo que necesito y también para poder agregar el método "terminar". Debe establecer la propiedad habilitada de la tarea periódica en False y guardarla antes de eliminarla. Toda la subclasificación no es obligatoria, el método schedule_every es el que realmente hace el trabajo. Cuando esté listo para finalizar su tarea (si no la subclasificó), puede usar PeriodicTask.objects.filter (name = ...) para buscar su tarea, deshabilitarla y luego eliminarla.

¡Espero que esto ayude!

from djcelery.models import PeriodicTask, IntervalSchedule
from datetime import datetime
class TaskScheduler(models.Model):
periodic_task = models.ForeignKey(PeriodicTask)
@staticmethod
def schedule_every(task_name, period, every, args=None, kwargs=None):
""" schedules a task by name every "every" "period". So an example call would be:
TaskScheduler(''mycustomtask'', ''seconds'', 30, [1,2,3])
that would schedule your custom task to run every 30 seconds with the arguments 1,2 and 3 passed to the actual task.
"""
permissible_periods = [''days'', ''hours'', ''minutes'', ''seconds'']
if period not in permissible_periods:
raise Exception(''Invalid period specified'')
# create the periodic task and the interval
ptask_name = "%s_%s" % (task_name, datetime.datetime.now()) # create some name for the period task
interval_schedules = IntervalSchedule.objects.filter(period=period, every=every)
if interval_schedules: # just check if interval schedules exist like that already and reuse em
interval_schedule = interval_schedules[0]
else: # create a brand new interval schedule
interval_schedule = IntervalSchedule()
interval_schedule.every = every # should check to make sure this is a positive int
interval_schedule.period = period
interval_schedule.save()
ptask = PeriodicTask(name=ptask_name, task=task_name, interval=interval_schedule)
if args:
ptask.args = args
if kwargs:
ptask.kwargs = kwargs
ptask.save()
return TaskScheduler.objects.create(periodic_task=ptask)
def stop(self):
"""pauses the task"""
ptask = self.periodic_task
ptask.enabled = False
ptask.save()
def start(self):
"""starts the task"""
ptask = self.periodic_task
ptask.enabled = True
ptask.save()
def terminate(self):
self.stop()
ptask = self.periodic_task
self.delete()
ptask.delete()

Answer 2

Hay una biblioteca llamada django-apio-beat que proporciona los modelos que uno necesita. Para hacer que cargue dinámicamente nuevas tareas periódicas, uno debe crear su propio Programador.

from django_celery_beat.schedulers import DatabaseScheduler
class AutoUpdateScheduler(DatabaseScheduler):
def tick(self, *args, **kwargs):
if self.schedule_changed():
print(''resetting heap'')
self.sync()
self._heap = None
new_schedule = self.all_as_schedule()
if new_schedule:
to_add = new_schedule.keys() - self.schedule.keys()
to_remove = self.schedule.keys() - new_schedule.keys()
for key in to_add:
self.schedule[key] = new_schedule[key]
for key in to_remove:
del self.schedule[key]
super(AutoUpdateScheduler, self).tick(*args, **kwargs)
@property
def schedule(self):
if not self._initial_read and not self._schedule:
self._initial_read = True
self._schedule = self.all_as_schedule()
return self._schedule

Answer 3

No, lo siento, esto no es posible con el celerybeat regular.

Pero es fácilmente extensible para hacer lo que quiera, por ejemplo, el planificador django-apio es solo una subclase leyendo y escribiendo el cronograma en la base de datos (con algunas optimizaciones en la parte superior).

También puedes usar el planificador django-apio incluso para proyectos que no son de Django.

Algo como esto:

• Instala django + django-apio:

  $ pip install -U django django-apio

• Agregue las siguientes configuraciones a su apioconfig:

  DATABASES = {
  ''default'': {
  ''NAME'': ''celerybeat.db'',
  ''ENGINE'': ''django.db.backends.sqlite3'',
  },
  }
  INSTALLED_APPS = (''djcelery'', )

• Crea las tablas de la base de datos:

  $ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig

• Comience apio de cebo con el programador de base de datos:

  $ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig /
  -S djcelery.schedulers.DatabaseScheduler

También está el comando djcelerymon que se puede usar para proyectos que no son de Django para iniciar apiocam y un servidor web Admin de Django en el mismo proceso, puede usarlo para editar también sus tareas periódicas en una bonita interfaz web:

 $ djcelerymon

(Tenga en cuenta que por alguna razón, djcelerymon no se puede detener usando Ctrl + C, tiene que usar Ctrl + Z + kill% 1)

Answer 4

Puede ver este flask-djcelery que configura matraz y djcelery y también proporciona api de búsqueda navegable