Как динамически добавлять/удалять периодические задания на сельдерей (celerybeat)

Ответ 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'].

Предположим, вы хотите отслеживать и изменять свой график ударов сельдерея с помощью внешнего приложения. Тогда у вас есть несколько вариантов:

1. Вы можете open файл базы данных полки и прочитать его содержимое как словарь. Напишите обратно в этот файл для внесения изменений.
2. Вы можете запустить другой экземпляр приложения Celery и использовать его для изменения файла полки, как описано выше.
3. Вы можете использовать пользовательский класс планировщика из django-celery-beat, чтобы сохранить расписание в базе данных, управляемой django, и получить доступ к записям там.
4. Вы можете использовать планировщик из celerybeat-mongo, чтобы сохранить расписание в бэкенде MongoDB и получить доступ к записям в нем.