Обновление задач в Сельдере с RabbitMQ - PullRequest
Новая
       14

Обновление задач в Сельдере с RabbitMQ

0 голосов
/ 25 марта 2019

Я использую Celery в своем проекте django для создания задач для отправки электронной почты в определенное время в будущем. Пользователь может создать экземпляр уведомления с полем notify_on datetime. Затем я передаю значение notify_on как eta. 

class Notification(models.Model):
    ...
    notify_on = models.DateTimeField()


def notification_post_save(instance, *args, **kwargs):
    send_notification.apply_async((instance,), eta=instance.notify_on)

signals.post_save.connect(notification_post_save, sender=Notification)

Проблема с этим подходом заключается в том, что если пользователь изменит notify_on, он получит два (или более) уведомления вместо одного.

Вопрос в том, как обновить задачу, связанную с определенным уведомлением, или каким-либо образом удалить старую и создать новую.

Ответы [ 2 ]

2 голосов
/ 25 марта 2019

Прежде всего, используя post_save, мы не можем получить старые данные. Итак, здесь я переопределяю метод save() модели Notification. Кроме того, создайте поле для хранения сельдерея task_id. 

<b>from celery.task.control import revoke</b>


class Notification(models.Model):
    ...
    notify_on = models.DateTimeField()
    <b>celery_task_id = models.CharField(max_length=100)

    def save(self, *args, **kwargs):
        pre_notify_on = Notification.objects.get(pk=self.pk).notify_on
        super().save(*args, **kwargs)
        post_notify_on = self.notify_on
        if not self.celery_task_id:  # initial task creation
            task_object = send_notification.apply_async((self,), eta=self.notify_on)
            Notification.objects.filter(pk=self.pk).update(celery_task_id=task_object.id)
        elif pre_notify_on != post_notify_on:
            # revoke the old task
            revoke(self.celery_task_id, terminate=True)
            task_object = send_notification.apply_async((self,), eta=self.notify_on)
            Notification.objects.filter(pk=self.pk).update(celery_task_id=task_object.id)</b>

Ссылки

  1. Отменить уже выполняемое задание с помощью Celery?
  2. Django: как получить доступ к оригинальному (неизмененному) экземпляру в сигнале post_save
0 голосов
/ 25 марта 2019

Я думаю, что нет необходимости удалять предыдущие задачи.Вам просто нужно подтвердить, что выполняемая задача является продолжительной.Для этого создайте новое поле с именем контрольной суммы, которое является полем UUID, обновляйте это поле каждый раз, когда вы меняете notify_on.Проверьте эту контрольную сумму в задаче, в которую вы отправляете электронное письмо. 

class Notification(models.Model):
    checksum = models.UUIDField(default=uuid.uuid4)
    notify_on = models.DateTimeField()

def notification_post_save(instance, *args, **kwargs):
    send_notification.apply_async((instance.id, str(instance.checksum)),eta=instance.notify_on)

signals.post_save.connect(notification_post_save, sender=Notification)


@shared_task 
def send_notification(notification_id, checksum):
    notification = Notification.objects.get(id=notification_id)
    if str(notification.checksum) != checksum:
        return False
    #send email

Также, пожалуйста, не отправляйте сигнал каждый раз при уведомлении об объекте, сохраните, просто отправьте его при изменении notify_on.Вы также можете проверить это Определить измененные поля в django post_save signal

...