Как создать отдельные задания для каждого объекта с сельдереем в моделях Django - PullRequest
Новая
       59

Как создать отдельные задания для каждого объекта с сельдереем в моделях Django

0 голосов
/ 14 декабря 2018

Предположим, у нас есть следующее поле модели: 

class Project(models.Model):
    project_name = models.CharField(max_length=200,unique=True)
    project_scan = models.IntegerField()  ### Scan interval
    project_status = models.BooleanField() ### To Enable "Scan" or Disable "Scan" Tasks

Предположим, у нас есть 2 объекта проекта: 

1. Project(project_name='test1',project_scan=5) ### Scan `test1` every `5` hour
2. Project(project_name='test2',project_scan=10) ### Scan `test2` every `10` hour

Tasks.py 

@task(name='project_tasks')
def Project_Tasks():
    get_all_projects = Project.objects.all()
    for each_project in get_all_project:
        if each_project.project_status == True: ### Checking if it "Scan" is allowed.
            get_interval = each_project.project_scan
            get_name = each_project.project_name
            print(get_name)

Мой вопрос: Как мне запускать задачи для каждого объекта на основе заданного интервала project_scan?Поскольку Celery beat принимает имя Задачи в качестве аргумента для выполнения сканирования, например: PeriodicTask.objects.create(interval=given_interval, name='I dont know', task='project_tasks', )

Как создать отдельный экземпляр для каждой задачи проекта?

Я пытался создать поле intervalSchedule в models.py, ноне сработало: 

class Project(models.Model):
    project_name = models.CharField(max_length=200,unique=True)
    project_scan = models.IntegerField()  ### Scan interval
    project_status = models.BooleanField() ### To Enable "Scan" or Disable "Scan" Tasks
    schedule = IntervalSchedule()

1 Ответ

0 голосов
/ 14 декабря 2018

Вы можете добавить сигналы использования: 

from django.db.models.signals import post_save
from django.dispatch import receiver

from django_celery_beat.models import PeriodicTask, IntervalSchedule


class Project(models.Model):
    project_name = models.CharField(max_length=200,unique=True)
    project_scan = models.IntegerField()  
    project_status = models.BooleanField()


    def set_periodic_task(self, task_name):
        schedule = self.get_or_create_interval()
        PeriodicTask.objects.create(
            interval=schedule, 
            name=f'{self.project_name}-{self.id}', 
            task=task_name,
        )

    def get_or_create_interval(self):
        schedule, created = IntervalSchedule.objects.get_or_create(
            every=self.project_scan,
            period=IntervalSchedule.HOURS,
        )
        return schedule

    def get_periodic_task(self, task_name):
        interval = self.get_or_create_interval()
        periodic_task = PeriodicTask.objects.get(
            interval=interval, 
            name=f'{self.project_name}-{self.id}', 
            task=task_name,
        )
        return periodic_task

    def sync_disable_enable_task(self, task_name):
        periodic_task = self.get_periodic_task(task_name)
        periodic_task.enabled = self.project_status
        periodic_task.save()


@receiver(post_save, sender=Project)
def set_or_sync_periodic_task(sender, instance=None, created=False, **kwargs):
    if created:
        instance.set_periodic_task(task_name='project_tasks')
    else:
        instance.sync_disable_enable_task(task_name='project_tasks')

Что у вас есть: при создании нового экземпляра Project новое Периодическое задание сохраняется с методом set_periodic_task.Если вы хотите disable или enable периодическое задание вашего экземпляра, просто измените статус project_status и сохраните его.Он включит sync_disable_enable_task метод для включения или отключения.

, если вы хотите передать аргументы, вы можете сделать: 

 PeriodicTask.objects.create(
     interval=schedule,
     name=f'{self.project_name}-{self.id}', 
     task='proj.tasks.import_contacts',
     args=json.dumps(['arg1', 'arg2']),
     kwargs=json.dumps({
        'some_kwarg': '123,
     }),
)
...