Django自定义信号实战：解决update操作监听缺失问题
本文深入解析Django信号机制，重点解决内置信号中update操作缺失的问题。通过定义自定义Signal、注册接收器及重写QuerySet update方法，实现模型数据变更时的自动触发逻辑。文章提供完整代码示例，涵盖信号定义、注册及回调处理，帮助开发者灵活扩展Django模型行为。
Django框架内置了多种信号（signals），旨在监听特定操作并向相关组件发送通知，从而实现应用模块间的解耦。
根据官方文档说明：
Django提供了一个“信号分发器”，允许应用解耦。当框架其他部分发生操作时，相关应用会收到通知。
简而言之，信号机制允许特定的发送者（sender）通知一组接收者（receiver）某些操作已发生。这在多处代码需要响应同一事件时非常有用。 Django内置了一些信号，定义在django/db/models/signal.py中，例如：
Model signals pre_init # django的modal执行其构造方法前，自动触发 post_init # django的modal执行其构造方法后，自动触发 pre_save # django的modal对象保存前，自动触发 post_save # django的modal对象保存后，自动触发 pre_delete # django的modal对象删除前，自动触发 post_delete # django的modal对象删除后，自动触发 m2m_changed # django的modal中使用m2m字段操作第三张（add,remove,clear）前后，自动触发 class_prepared # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发 Management signals pre_migrate # 执行migrate命令前，自动触发 post_migrate # 执行migrate命令后，自动触发 Request/response signals request_started # 请求到来前，自动触发 request_finished # 请求结束后，自动触发 got_request_exception # 请求异常后，自动触发 Test signals setting_changed # 使用test测试修改配置文件时，自动触发 template_rendered # 使用test测试渲染模板时，自动触发 Database Wrappers connection_created # 创建数据库连接时，自动触发
使用示例如下：
利用这些内置信号可以实现模型中的联动操作。例如，若需记录通过模型更新记录时的操作者日志，可以直接在相关代码处使用post_save装饰器：
或者改写post_save逻辑以记录相关信息，从而一劳永逸地解决日志记录问题。此外，也可以在请求（request）处理阶段记录请求信息。
from django.core.signals import request_finished
from django.dispatch import receiver
@receiver(request_finished)
def my_callback(sender, **kwargs):
print("Request finished!")
那么，如何自定义信号（signals）呢？
a. 定义信号文件
import django.dispatch pizza_done = django.dispatch.Signal(providing_args=["toppings", "size"])
b. 注册信号
def callback(sender, **kwargs):
print("callback")
print(sender,kwargs)
pizza_done.connect(callback)
c. 触发信号
from 路径 import pizza_done pizza_done.send(sender='seven',toppings=123, size=456)
具体需求场景：
项目中有一个需求，当模型（即数据库数据）被修改或删除时，自动触发一个Redis同步任务（后来发现该需求并无实际意义……）。模型保存时有post_save，删除时有post_delete，唯独缺少update相关的信号。由于代码中频繁使用update场景，因此查阅了为何官方不提供update信号的原因。
参考链接：https://code.djangoproject.com/ticket/12184
其实很早就有人向Django官方提议添加此类功能。至于为何未被官方接受，可参考相关讨论。当时Django 1.9版本仍不支持此功能，因此只能自行编写实现，若后续官方支持或发现有问题，再行撤销。
解决方案如下：
signals.py文件内容：
# coding:utf-8 from django.dispatch import Signal post_update = Signal(providing_args=["user"])
models.py文件内容：
-----------针对特定模型，重写其QuerySet中的update方法-----------
//引入自定义的signal文件
from tools import signals
class MyCustomQuerySet(models.query.QuerySet):
def update(self, **kwargs):
super(MyCustomQuerySet, self).update(**kwargs)
//update被调用时， 发送该singalsignals
signals.post_update.send(sender=self.model, user="xxx")
print("finished!")
class MyCustomManager(models.Manager):
def get_queryset(self):
return MyCustomQuerySet(self.model, using=self._db)
class crontab_ping(models.Model):
name = models.CharField(max_length=64, blank=True, null=True)
objects = MyCustomManager()
callback.py文件内容：
-------接收信号，触发相应操作----------
from tools.signals import post_update
@receiver(post_update)
def post_update_callback(sender, **kwargs):
print(kwargs['user'])
print("post_update_success")
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
