Django Celery. Пост Publush на publish_date

У меня есть модель Post. В ней есть поле publish_date. Если пост имеет статус planned, то мне нужно выполнить функцию публикации по дате публикации (publish_date). Как я могу это сделать?

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

Я пытался сделать это, но ничего не вышло, задание не выполнено.

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()

Вы можете попробовать установить модуль django-celery-results, который поможет вам просматривать журналы Celery из панели администратора.

@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})

Кроме того, убедитесь, что и рабочий, и ритм запущены.

Я бы посоветовал не работать с задачей для установки статуса. Представьте, что вы позже измените дату публикации, вам нужно будет отменить предыдущую задачу и запланировать новую, что усложнит задачу. Можно даже установить дату публикации в прошлом, и тогда задача может не выполниться.

Вы можете отфильтровать неопубликованные элементы. Действительно, мы можем определить модель следующим образом:

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)

и в ListView работе с:

from django.db.models.functions import Now


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

Таким образом, если в будущем вы установите publish_date, Post автоматически перестанет отображаться.

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.

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