如何使用 Django 在 Celery 中正确创建任务？答案

【问题标题】：How to create tasks correctly in Celery with Django?如何使用 Django 在 Celery 中正确创建任务？
【发布时间】：2021-09-27 18:33:59
【问题描述】：

我请求帮助完成这项任务。有一个通知模型。我想创建一个异步任务来创建通知。但我得到一个错误 MPTTModelBase 类型的对象不是 JSON 可序列化的。模型.py

class Comment(MPTTModel):
    """Модель комментариев"""
    content_type = models.ForeignKey(ContentType, verbose_name=_('Тип контента'), related_name='content_ct_%(class)s', on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField(verbose_name=_('ID контента'), db_index=True)
    content_object = GenericForeignKey('content_type', 'object_id')
    """Род.коммент"""
    parent = TreeForeignKey('self', on_delete=models.CASCADE, verbose_name=_('Родительский комментарий'), blank=True,
                            null=True, related_name='children')
    """Инфо, привязка, модерация"""
    content = models.TextField(verbose_name=_('Комментарий'))
    created_by = models.ForeignKey(AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='comment_created_by', verbose_name=_('Автор комментария'))
    is_published = models.BooleanField(verbose_name=_('Опубликовать'), default=True)
    time_create = models.DateTimeField(auto_now_add=True, verbose_name=_('Дата добавления'))
    """Generic FK"""
    rating = GenericRelation('Rating', related_query_name='%(class)s')
    notification = GenericRelation('Notification', related_query_name='%(class)s')


  def save(self, *args, **kwargs):
    send_create_notification.delay(self, 3)
    super().save(*args, **kwargs)

services.py

def create_notification(instance, type):
"""Notification create"""
from modules.core.models import Notification
model_object = instance
obj = model_object.content_object
try:
    text = model_object.content[0:120]
except:
    text = None
try:
    to_user = obj.created_by
except:
    to_user = obj
from_user = model_object.created_by
now = timezone.now()
last_minute = now - datetime.timedelta(seconds=60)
similar_actions = Notification.objects.filter(from_user=from_user, to_user=from_user, type=type, time_create__gte=last_minute)
if obj:
    from django.contrib.contenttypes.models import ContentType
    content_type = ContentType.objects.get_for_model(obj)
    similar_actions = similar_actions.filter(content_type=content_type, object_id=obj.id)
if not similar_actions:
    if text:
        notification = Notification(from_user=from_user, to_user=to_user, type=type, content_object=obj, text=text)
    else:
        notification = Notification(from_user=from_user, to_user=to_user, type=type, content_object=obj)
    notification.save()
    return True
return False

tasks.py

  @shared_task()
    def send_create_notification(self, type):
        return create_notification(self, type)

【问题讨论】：

标签： python django redis celery django-celery

【解决方案1】：

send_create_notification.delay(self, 3) 将尝试序列化您的 Comment 实例，它不是 JSON 可序列化的。您可以使用pickle serializer，但不建议这样做。

我建议您将评论 ID 作为参数发送给任务：

send_create_notification.delay(self.pk, 3)

并获取任务上的实例

    @shared_task()
    def send_create_notification(comment_id, type):
        comment = Comment.objects.get(pk=comment_id)
        return create_notification(comment, type)

【讨论】：

太棒了！我忘了你可以用 id 做任何事情！
我这样通过 send_create_notification_task.delay(self.__class__.__name__, self.pk, 3)，通过条件 def send_create_notification_task(model, obj_id, type): instance = model.objects.get (pk=obj_id) 我得到错误 AttributeError: 'str' object has no attribute 'objects'
@Razilator self.__class__.__name__ 是一个字符串，就像写"Comment".objects.get(.. 你需要dinamically load the model using get_model