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

Question

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

我的应用是这样设置的。

我的博客模型：

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)

我的信号.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',)

Answer 1

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

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

Answer 2

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

你写道：

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)

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

1. 使用 @shared_task 装饰器而不使用相应的导入 from celery import shared_task
2. 错误的意图 - 没有理由打算使用该功能
3. 在类外使用 self 参数

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