Update tasks in Celery with RabbitMQ

Question

I'm using Celery in my django project to create tasks to send email at a specific time in the future. User can create a Notification instance with notify_on datetime field. Then I pass value of notify_on as a 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)

The problem with that approach is that if notify_on will be changed by the user, he will get two(or more) notifications instead of one.

The question is how do I update the task associated with a specific notification, or somehow delete the old one and create new.

JPG · Accepted Answer · 2019-03-25T13:01:37.083

First of all, by using post_save, we can't fetch the old data. So, here I'm overriding the save() method of the Notification model. Apart from that,create a field to store the celery task_id.

from celery.task.control import revoke


class Notification(models.Model):
    ...
    notify_on = models.DateTimeField()
    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)

Reference

Ok, but how will it help me to update `eta` on the task if the model instance was updated? And notification does not get send right after the creation of the task, it will be sent in some point in the future. — Nikita Tonkoskur, Mar 25 '19 at 12:25
It worked, thanks! I just had to try - except the first line of the `save` method, it gave me `DoesNotExist Notification matching query does not exist.` on the iniatial creation. Really appreciate your help! — Nikita Tonkoskur, Mar 25 '19 at 13:42

score 0 · Answer 2 · answered Mar 25 '19 at 12:20

0

I think there is no need for deleting the previous tasks. You just have to validate that the task that is executing is the lasted one. For that create a new field called checksum that is a UUID field update that field everytime you change notify_on. Check this checksum in the task where you are sending the email.

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

Also please don't send signal everytime on notification object save just send this when notify_on changes. You can also check this Identify the changed fields in django post_save signal

answered Mar 25 '19 at 12:20

Siddharth Jain

21
4

That doesn't work. Let's say user created notification with `notify_on`=14:40, task gets created immediatly and will execute and that time. If later, before the task executed, user would update notification and set `notify_on` to 14:50 that task would and should register too. User will get two notifications instead of one. – Nikita Tonkoskur Mar 25 '19 at 12:49
No this will not happen as during the task execution I am calling the db and getting the latest checksum. Let's say user created notification with notify_on=14:40 task gets registered with checksum "123"(assume). I now a user changed the notify_on to 14:50 new task is created with different checksum "456". Now when the first task is executed at 14:40 it will check the latest checksum from db i.e "456" with the checksum that is sent with the task args i.e "123" and it will return False. – Siddharth Jain Mar 26 '19 at 06:58
But I think JPG solution is better to just revoke the task. – Siddharth Jain Mar 26 '19 at 07:00
Oh, now I get it. Thank you, I’ll definitely use this technique later – Nikita Tonkoskur Mar 26 '19 at 07:04

Update tasks in Celery with RabbitMQ

2 Answers2

Reference

Linked