创建模型对象时不会触发 Django 信号

如何解决创建模型对象时不会触发 Django 信号

我正在使用 Django 信号触发任务（使用 django celery 包向订阅者发送大量电子邮件），当管理员发布博客文章时，由 Django 管理员创建。但是没有触发信号。我在不打印的信号中有一个打印语句，即在创建新博客后未收到信号。

我的应用是这样设置的。

我的博客模型：

class BlogPost(models.Model):

    author = models.CharField(max_length=64,default='Admin')    
    image = models.ImageField(blank=True,null=True)
    title = models.CharField(max_length=255)
    .................../

我的任务文件

from django.core.mail import send_mail

from travel_crm.settings import EMAIL_HOST_USER

    @shared_task
    def send_mails(self,*args,**kwargs):
    
        subscribers = self.kwargs['subscribers']
        blog = self.kwargs['blog']
        for abc in subscribers:
            emailad = abc.email
            send_mail('New Blog Post ',f" Checkout our new blog with title {blog.title} ",EMAIL_HOST_USER,[emailad],fail_silently=False)

我的signals.py文件

from .tasks import send_mails
from apps.blogs.models import BlogPost,Subscribers
from django.db.models.signals import post_save
from django.dispatch import receiver


    @receiver(post_save,sender=BlogPost)
    def email_task(sender,instance,created,**kwargs):
        print(123456789)
        if created:
            print(123456789)
            subscribers = Subscribers.objects.all()
            blog = BlogPost.objects.latest('date_created')
            print(blog)
            # task = send_mails(subscribers,blog)
            # task.delay()
            send_mails.delay(subscribers,blog)

我的初始化文件

from __future__ import absolute_import,unicode_literals
from apps.blogs.celery_files.celery import app as celery_app
__all__ = ('celery_app',)

解决方法

您必须在某处导入您的 signals.py 才能注册，django 文档建议在应用程序的 ready() 文件内的 apps.py 方法中导入它，如下所示：

from django.apps import AppConfig


class BlogsConfig(AppConfig):
    name = 'blogs'

    def ready(self):
        import blogs.celery_files.signals

https://docs.djangoproject.com/en/3.2/topics/signals/#connecting-receiver-functions

检查您的我的任务文件文件。您是否共享了整个文件？

你写道：

from django.core.mail import send_mail

from travel_crm.settings import EMAIL_HOST_USER

    @shared_task
    def send_mails(self,*args,**kwargs):
    
        subscribers = self.kwargs['subscribers']
        blog = self.kwargs['blog']
        for abc in subscribers:
            emailad = abc.email
            send_mail('New Blog Post ',f" Checkout our new blog with title {blog.title} ",EMAIL_HOST_USER,[emailad],fail_silently=False)

如果这是整个文件，则您有多个问题需要修复：

在没有相应导入的情况下使用 @shared_task 装饰器 from celery import shared_task
错误的意图 - 没有理由预期该功能
在类外使用 self 参数

您在signals.py 中有同样的意图问题。