Django信号
1、基本概念Django 信号是一种观察者模式的实现，用于在框架内部或应用之间传递信息，实现解耦的组件通信。当特定事件发生时（如模型保存、删除等），信号系统会通知所有已注册的接收器。
组成：
[*]发送者 (Sender)：触发信号的对象，通常是模型类，但也可以是任何 Python 对象。
[*]信号 (Signal)：事件本身的载体，是 django.dispatch.Signal的实例。
[*]接收者 (Receiver)：响应信号的函数或方法，在信号发送时被调用。
2、内置信号
2.1 模型信号
from django.db.models.signals import (
pre_init, post_init, # 模型实例化前后
pre_save, post_save, # 模型保存前后
pre_delete, post_delete, # 模型删除前后
m2m_changed, # 多对多关系变更
class_prepared # 模型类准备就绪
)2.2 请求/响应信号
from django.core.signals import (
request_started, # 请求开始
request_finished, # 请求结束
got_request_exception # 请求异常
)2.3 数据库包装器信号
from django.db.backends.signals import (
connection_created # 数据库连接创建
)3、源码分析
3.1 Signal简化版代码
class Signal:
def __init__(self, use_caching=False):
# 接收者列表
self.receivers = []
# 是否使用缓存
self.use_caching = use_caching
# 用于优化的缓存
self.sender_receivers_cache = weakref.WeakKeyDictionary() if use_caching else {}
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
"""注册接收器"""
# 获取或创建dispatch_uid
if dispatch_uid:
lookup_key = (dispatch_uid, _make_id(sender))
else:
lookup_key = (_make_id(receiver), _make_id(sender))
# 如果是弱引用，使用weakref
if weak:
ref = weakref.ref
receiver_object = receiver
if hasattr(receiver, "__self__") and hasattr(receiver, "__func__"):
ref = weakref.WeakMethod
receiver_object = receiver.__self__
receiver = ref(receiver)
weakref.finalize(receiver_object, self._remove_receiver)
# 将接收器添加到列表
self.receivers.append((lookup_key, receiver, is_async))
def disconnect(self, receiver=None, sender=None, dispatch_uid=None):
"""
断开接收者与信号的连接, 清理缓存和self.receivers
"""
if dispatch_uid:
lookup_key = (dispatch_uid, _make_id(sender))
else:
lookup_key = (_make_id(receiver), _make_id(sender))
disconnected = False
with self.lock:
self._clear_dead_receivers()
for index in range(len(self.receivers)):
r_key, *_ = self.receivers
if r_key == lookup_key:
disconnected = True
del self.receivers
break
self.sender_receivers_cache.clear()
return disconnected
def has_listeners(self, sender=None):
sync_receivers, async_receivers = self._live_receivers(sender)
return bool(sync_receivers) or bool(async_receivers)
def send(self, sender, **named):
"""
发送信号给所有连接的接收者
"""
sync_receivers = self._live_receivers(sender)
for receiver in sync_receivers:
# 调用接收器
response = receiver(signal=self, sender=sender, **named)
responses.append((receiver, response))
return responses
def send_robust(self, sender, **named):
"""
发送信号，即使某些接收者抛出异常也继续执行
"""
pass
def _live_receivers(self, sender):
"""
获取活动的接收者
"""
pass3.2 调用流程
class ModelBase(type):
"""Model的元类，控制Model类的创建"""
def __new__(cls, name, bases, attrs, **kwargs):
pass
def _prepare(cls):
# ...
# 模型类准备就绪
class_prepared.send(sender=cls)
class Model(AltersData, metaclass=ModelBase):
def __init__(self, *args, **kwargs):
# 模型实例化前
pre_init.send(sender=cls, args=args, kwargs=kwargs)
# ...
# 模型实例化后
post_init.send(sender=cls, instance=self)
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
if not meta.auto_created:
# 模型保存前
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# ...
if not meta.auto_created:
# 模型保存后
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
save_base.alters_data = True4、基本使用
4.1 模型保存信号
from django.db.models.signals import pre_save, post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from myapp.models import Profile
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
"""
在用户创建时自动创建用户档案
"""
if created:
Profile.objects.create(user=instance)
@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
"""
保存用户档案
"""
instance.profile.save()
@receiver(pre_save, sender=User)
def pre_save_user_handler(sender, instance, **kwargs):
"""
在用户保存前执行操作
"""
# 例如：自动生成用户名
if not instance.username:
instance.username = f"user_{instance.email.split('@')}"4.2 请求信号
from django.core.signals import request_started, request_finished
from django.dispatch import receiver
@receiver(request_started)
def request_started_handler(sender, environ, **kwargs):
"""
请求开始时记录日志
"""
import logging
logger = logging.getLogger('django.request')
logger.info(f"Request started: {environ.get('PATH_INFO')}")
@receiver(request_finished)
def request_finished_handler(sender, **kwargs):
"""
请求结束时记录日志
"""
import logging
logger = logging.getLogger('django.request')
logger.info("Request finished")5、自定义信号
5.1 定义信号
# myapp/signals.py
from django.dispatch import Signal
# 定义自定义信号, providing_args, 声明信号发送时传递的参数列表, 这些参数会在信号发送时作为关键字参数传递给接收者
user_registered = Signal(providing_args=["user", "request"])
user_logged_in = Signal(providing_args=["user", "request"])5.2 发送自定义信号
# myapp/views.py
from django.contrib.auth import login
from django.shortcuts import render, redirect
from .forms import UserRegistrationForm
from .signals import user_registered
def register_view(request):
if request.method == 'POST':
form = UserRegistrationForm(request.POST)
if form.is_valid():
user = form.save()
# 发送用户注册信号
user_registered.send(
sender=user.__class__,
user=user,
request=request
)
login(request, user)
return redirect('home')
else:
form = UserRegistrationForm()
return render(request, 'registration/register.html', {'form': form})5.3 接收自定义信号
# myapp/handlers.py
from django.dispatch import receiver
from django.core.mail import send_mail
from django.conf import settings
from .signals import user_registered
@receiver(user_registered)
def send_welcome_email(sender, user, request, **kwargs):
"""
发送欢迎邮件给新注册用户
"""
subject = 'Welcome to Our Site!'
message = f'Hello {user.username}, thank you for registering!'
from_email = settings.DEFAULT_FROM_EMAIL
recipient_list =
send_mail(subject, message, from_email, recipient_list)
@receiver(user_registered)
def create_user_profile(sender, user, request, **kwargs):
"""
创建用户档案
"""
from .models import UserProfile
UserProfile.objects.create(user=user)6、最佳实践
# myapp/signals.py - 定义所有信号
from django.dispatch import Signal
user_registered = Signal(providing_args=["user", "request"])
order_created = Signal(providing_args=["order", "user"])
# myapp/handlers.py - 定义所有信号处理器
from django.dispatch import receiver
from .signals import user_registered, order_created
@receiver(user_registered)
def handle_user_registered(sender, user, request, **kwargs):
pass
@receiver(order_created)
def handle_order_created(sender, order, user, **kwargs):
pass
# myapp/__init__.py - 导入信号处理器以确保它们被注册
default_app_config = 'myapp.apps.MyAppConfig'
# myapp/apps.py - 在应用配置中导入信号
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = 'myapp'
def ready(self):
# 导入信号处理器
import myapp.handlers
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ 过来提前占个楼
页:
[1]