Django Celery. Publush post on publish_date

I have a Post model. There is a field publish_date. If the post has a planned status, then i need to perform the publishing function on the publish_date. How can I do this?

models.py:

class Post(models.Model):
    STATES = (
        ('draft', 'Draft'),
        ('published', 'Published'),
        ('planned', 'Planned')
    )
    state = models.CharField(choices=STATES, default=STATES[0][0])
    channels = models.ManyToManyField('channel.Channel')
    creator = models.ForeignKey('authentication.User', on_delete=models.SET_NULL, null=True)
    publish_date = models.DateTimeField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)


@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, **kwargs):
    # There should be a task setting for a specific time here, as I understand it

I tried to do this, but it didn't work, the task is not completed.

models.py

@receiver(post_save, sender=Post)
def reschedule_publish_task(sender, instance, **kwargs):
    task_id = f"publish_post_{instance.id}"

    if instance.state == 'planned' and instance.publish_date:
        publish_post.apply_async((instance.id,), eta=instance.publish_date, task_id=task_id)

tasks.py

@shared_task
def publish_post(post_id: int) -> None:
    from .models import Post

    post = Post.objects.filter(id=post_id).first()
    if post:
        if post.state == 'planned' and post.publish_date <= now():
            post.state = 'published'
            post.save()

You can try installing the django-celery-results module, which will help you view Celery logs from the admin panel.

@shared_task
def publish_post(post_id: int) -> None:
   from .models import Post

   post = Post.objects.filter(id=post_id).first()
   if post:
       if post.state == 'planned' and post.publish_date <= now():
           post.state = 'published'
           post.save()
           return JsonResponse({"success": True})
      return JsonResponse({"success": False, "error": "date of post Error", "data": f"Post ID: {post_id} \n Post date: {post.publish_date} \n Post state {post.state} \n Date now: {now()}"})
   return JsonResponse({"success": False, "error": "Post not found", "data": post_id})

Also, make sure that both the worker and beat are running.

I would advise not to work with a task to set the status. Imagine that you later change the publish date, you need to cancel the previous task and schedule a new one, making it more complicated. One could even set the publish date in the past, and then the task might not run.

You can filter out unpublished elements. Indeed, we can define the model with:

from django.conf import settings


class Post(models.Model):
    STATES = (('draft', 'Draft'), ('planned', 'Planned'))
    state = models.CharField(choices=STATES, default=STATES[0][0])
    channels = models.ManyToManyField('channel.Channel')
    creator = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True
    )
    publish_date = models.DateTimeField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

and in the ListView work with:

from django.db.models.functions import Now


class PostListView(models.Model):
    queryset = Post.object.filter(state='planned', publish_date__lte=Now())

That way, if you would later set the publish_date in the future, the Post automatically no longer is shown.

Note: It is normally better to make use of the settings.AUTH_USER_MODEL [Django-doc] to refer to the user model, than to use the User model [Django-doc] directly. For more information you can see the referencing the User model section of the documentation.

Вернуться на верх