Как динамически добавлять/удалять периодические задания на сельдерей (celerybeat)
Если у меня есть функция, определенная следующим образом:
def add(x,y):
return x+y
Есть ли способ динамически добавлять эту функцию в качестве циклической периодики для сельдерея и запускать ее во время выполнения? Я хотел бы иметь возможность делать что-то вроде (псевдокода):
some_unique_task_id = celery.beat.schedule_task(add, run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
Я также хотел бы остановить или удалить эту задачу динамически с помощью чего-то вроде (псевдокода):
celery.beat.remove_task(some_unique_task_id)
или
celery.beat.stop(some_unique_task_id)
FYI Я не использую djcelery, который позволяет вам управлять периодическими задачами с помощью администратора django.
Ответы
Ответ 1
Нет, извините, это невозможно с обычным сельдереем.
Но он легко расширяется, чтобы делать то, что вы хотите, например. джанго-сельдерей
scheduler - это просто чтение подкласса и запись расписания в базу данных
(с некоторыми оптимизациями сверху).
Также вы можете использовать планировщик django-celery даже для проектов, отличных от Django.
Что-то вроде этого:
-
Установите django + django-celery:
$pip install -U django django-celery
-
Добавьте в свой celeryconfig следующие параметры:
DATABASES = {
'default': {
'NAME': 'celerybeat.db',
'ENGINE': 'django.db.backends.sqlite3',
},
}
INSTALLED_APPS = ('djcelery', )
-
Создайте таблицы базы данных:
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
-
Запустите celerybeat с планировщиком базы данных:
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Также существует команда djcelerymon
, которая может использоваться для проектов, отличных от Django
для запуска celerycam и веб-сервера Django Admin в том же процессе, вы можете
используйте это, чтобы также редактировать свои периодические задачи в хорошем веб-интерфейсе:
$ djcelerymon
(Обратите внимание, что по какой-либо причине djcelerymon нельзя остановить с помощью Ctrl + C, вы
нужно использовать Ctrl + Z + kill% 1)
Ответ 2
На этот вопрос был дан ответ группы google.
Я НЕ АВТОР, все заслуги Жан-Марка
Здесь правильное решение для этого. Подтвержденная работа. По моему сценарию, Я подклассифицировал "Периодическую задачу" и создал из нее модель, поскольку я могу добавьте другие поля в модель, как мне нужно, и поэтому я мог бы добавить "прекратить" метод. Вы должны настроить периодическую задачу свойство False и сохраните его перед удалением. Целый подклассификация не обязательна, метод sched_every - тот, который действительно делает работу. Когда вы будете готовы завершить свою задачу (если вы не подклассифицировал его), вы можете просто использовать PeriodicTask.objects.filter(name=...) для поиска вашей задачи, отключите затем удалите его.
Надеюсь, это поможет!
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()
Ответ 3
Существует библиотека под названием django-celery-beat, которая предоставляет модели, которые вам нужны. Чтобы динамически загружать новые периодические задачи, нужно создать собственный планировщик.
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
Ответ 4
Вы можете проверить этот flask-djcelery, который настраивает флягу и djcelery, а также предоставляет доступный для просмотра api
Ответ 5
Это наконец стало возможным благодаря исправлению, включенному в celery v4.1.0. Теперь вам просто нужно изменить записи расписания в базе данных базы данных, и celery-beat будет действовать в соответствии с новым расписанием.
Документы смутно описывают, как это работает. Планировщик по умолчанию для сельдерея, PersistentScheduler
, использует файл полки в качестве базы данных расписаний. Любые изменения в словаре beat_schedule
в экземпляре PersistentScheduler
синхронизируются с этой базой данных (по умолчанию каждые 3 минуты) и наоборот. Документы описывают, как добавлять новые записи в beat_schedule
используя app.add_periodic_task
. Чтобы изменить существующую запись, просто добавьте новую запись с тем же name
. Удалите запись как из словаря: del app.conf.beat_schedule['name']
.
Предположим, вы хотите отслеживать и изменять свой график ударов сельдерея с помощью внешнего приложения. Тогда у вас есть несколько вариантов:
- Вы можете
open
файл базы данных полки и прочитать его содержимое как словарь. Напишите обратно в этот файл для внесения изменений. - Вы можете запустить другой экземпляр приложения Celery и использовать его для изменения файла полки, как описано выше.
- Вы можете использовать пользовательский класс планировщика из django-celery-beat, чтобы сохранить расписание в базе данных, управляемой django, и получить доступ к записям там.
- Вы можете использовать планировщик из celerybeat-mongo, чтобы сохранить расписание в бэкенде MongoDB и получить доступ к записям в нем.