Django信号

1、基本概念

Django 信号是一种观察者模式的实现，用于在框架内部或应用之间传递信息，实现解耦的组件通信。当特定事件发生时（如模型保存、删除等），信号系统会通知所有已注册的接收器。
组成：

• 发送者 (Sender)：触发信号的对象，通常是模型类，但也可以是任何 Python 对象。
  
• 信号 (Signal)：事件本身的载体，是 django.dispatch.Signal的实例。
  
• 接收者 (Receiver)：响应信号的函数或方法，在信号发送时被调用。
  

2、内置信号

2.1 模型信号

1. from django.db.models.signals import (
2.     pre_init, post_init,          # 模型实例化前后
3.     pre_save, post_save,          # 模型保存前后
4.     pre_delete, post_delete,      # 模型删除前后
5.     m2m_changed,                  # 多对多关系变更
6.     class_prepared                # 模型类准备就绪
7. )

复制代码

2.2 请求/响应信号

1. from django.core.signals import (
2.     request_started,              # 请求开始
3.     request_finished,             # 请求结束
4.     got_request_exception         # 请求异常
5. )

复制代码

2.3 数据库包装器信号

1. from django.db.backends.signals import (
2.     connection_created            # 数据库连接创建
3. )

复制代码

3、源码分析

3.1 Signal简化版代码

1. class Signal:
3.     def __init__(self, use_caching=False):
4.         # 接收者列表
5.         self.receivers = []
6.         # 是否使用缓存
7.         self.use_caching = use_caching
8.         # 用于优化的缓存
9.         self.sender_receivers_cache = weakref.WeakKeyDictionary() if use_caching else {}
11.     def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
12.         """注册接收器"""
13.         # 获取或创建dispatch_uid
14.         if dispatch_uid:
15.             lookup_key = (dispatch_uid, _make_id(sender))
16.         else:
17.             lookup_key = (_make_id(receiver), _make_id(sender))
19.         # 如果是弱引用，使用weakref
20.         if weak:
21.             ref = weakref.ref
22.             receiver_object = receiver
23.             if hasattr(receiver, "__self__") and hasattr(receiver, "__func__"):
24.                 ref = weakref.WeakMethod
25.                 receiver_object = receiver.__self__
26.             receiver = ref(receiver)
27.             weakref.finalize(receiver_object, self._remove_receiver)
29.         # 将接收器添加到列表
30.         self.receivers.append((lookup_key, receiver, is_async))
32.     def disconnect(self, receiver=None, sender=None, dispatch_uid=None):
33.         """
34.         断开接收者与信号的连接, 清理缓存和self.receivers
35.         """
36.         if dispatch_uid:
37.             lookup_key = (dispatch_uid, _make_id(sender))
38.         else:
39.             lookup_key = (_make_id(receiver), _make_id(sender))
41.         disconnected = False
42.         with self.lock:
43.             self._clear_dead_receivers()
44.             for index in range(len(self.receivers)):
45.                 r_key, *_ = self.receivers[index]
46.                 if r_key == lookup_key:
47.                     disconnected = True
48.                     del self.receivers[index]
49.                     break
50.             self.sender_receivers_cache.clear()
51.         return disconnected
53.     def has_listeners(self, sender=None):
54.         sync_receivers, async_receivers = self._live_receivers(sender)
55.         return bool(sync_receivers) or bool(async_receivers)
57.     def send(self, sender, **named):
58.         """
59.         发送信号给所有连接的接收者
60.         """
61.         sync_receivers = self._live_receivers(sender)
62.         for receiver in sync_receivers:
63.             # 调用接收器
64.             response = receiver(signal=self, sender=sender, **named)
65.             responses.append((receiver, response))
66.         return responses
68.    
70.     def send_robust(self, sender, **named):
71.         """
72.         发送信号，即使某些接收者抛出异常也继续执行
73.         """
74.         pass
77.     def _live_receivers(self, sender):
78.         """
79.         获取活动的接收者
80.         """
81.         pass

复制代码

3.2 调用流程

1. class ModelBase(type):
2.     """Model的元类，控制Model类的创建"""
4.     def __new__(cls, name, bases, attrs, **kwargs):
5.         pass
7.         
8.     def _prepare(cls):
9.         # ...
10.         # 模型类准备就绪
11.         class_prepared.send(sender=cls)
13. class Model(AltersData, metaclass=ModelBase):
14.     def __init__(self, *args, **kwargs):
15.         # 模型实例化前
16.         pre_init.send(sender=cls, args=args, kwargs=kwargs)
17.         # ...
18.         # 模型实例化后
19.         post_init.send(sender=cls, instance=self)
21.     def save_base(
22.         self,
23.         raw=False,
24.         force_insert=False,
25.         force_update=False,
26.         using=None,
27.         update_fields=None,
28.     ):
30.         if not meta.auto_created:
31.             # 模型保存前
32.             pre_save.send(
33.                 sender=origin,
34.                 instance=self,
35.                 raw=raw,
36.                 using=using,
37.                 update_fields=update_fields,
38.             )
39.         # ...
40.         if not meta.auto_created:
41.             # 模型保存后
42.             post_save.send(
43.                 sender=origin,
44.                 instance=self,
45.                 created=(not updated),
46.                 update_fields=update_fields,
47.                 raw=raw,
48.                 using=using,
49.             )
51.     save_base.alters_data = True

复制代码

4、基本使用

4.1 模型保存信号

1. from django.db.models.signals import pre_save, post_save
2. from django.dispatch import receiver
3. from django.contrib.auth.models import User
4. from myapp.models import Profile
6. @receiver(post_save, sender=User)
7. def create_user_profile(sender, instance, created, **kwargs):
8.     """
9.     在用户创建时自动创建用户档案
10.     """
11.     if created:
12.         Profile.objects.create(user=instance)
14. @receiver(post_save, sender=User)
15. def save_user_profile(sender, instance, **kwargs):
16.     """
17.     保存用户档案
18.     """
19.     instance.profile.save()
21. @receiver(pre_save, sender=User)
22. def pre_save_user_handler(sender, instance, **kwargs):
23.     """
24.     在用户保存前执行操作
25.     """
26.     # 例如：自动生成用户名
27.     if not instance.username:
28.         instance.username = f"user_{instance.email.split('@')[0]}"

复制代码

4.2 请求信号

1. from django.core.signals import request_started, request_finished
2. from django.dispatch import receiver
4. @receiver(request_started)
5. def request_started_handler(sender, environ, **kwargs):
6.     """
7.     请求开始时记录日志
8.     """
9.     import logging
10.     logger = logging.getLogger('django.request')
11.     logger.info(f"Request started: {environ.get('PATH_INFO')}")
13. @receiver(request_finished)
14. def request_finished_handler(sender, **kwargs):
15.     """
16.     请求结束时记录日志
17.     """
18.     import logging
19.     logger = logging.getLogger('django.request')
20.     logger.info("Request finished")

复制代码

5、自定义信号

5.1 定义信号

1. # myapp/signals.py
2. from django.dispatch import Signal
4. # 定义自定义信号, providing_args, 声明信号发送时传递的参数列表, 这些参数会在信号发送时作为关键字参数传递给接收者
5. user_registered = Signal(providing_args=["user", "request"])
6. user_logged_in = Signal(providing_args=["user", "request"])

复制代码

5.2 发送自定义信号

1. # myapp/views.py
2. from django.contrib.auth import login
3. from django.shortcuts import render, redirect
4. from .forms import UserRegistrationForm
5. from .signals import user_registered
7. def register_view(request):
8.     if request.method == 'POST':
9.         form = UserRegistrationForm(request.POST)
10.         if form.is_valid():
11.             user = form.save()
12.             
13.             # 发送用户注册信号
14.             user_registered.send(
15.                 sender=user.__class__,
16.                 user=user,
17.                 request=request
18.             )
19.             
20.             login(request, user)
21.             return redirect('home')
22.     else:
23.         form = UserRegistrationForm()
24.    
25.     return render(request, 'registration/register.html', {'form': form})

复制代码

5.3 接收自定义信号

1. # myapp/handlers.py
2. from django.dispatch import receiver
3. from django.core.mail import send_mail
4. from django.conf import settings
5. from .signals import user_registered
7. @receiver(user_registered)
8. def send_welcome_email(sender, user, request, **kwargs):
9.     """
10.     发送欢迎邮件给新注册用户
11.     """
12.     subject = 'Welcome to Our Site!'
13.     message = f'Hello {user.username}, thank you for registering!'
14.     from_email = settings.DEFAULT_FROM_EMAIL
15.     recipient_list = [user.email]
16.    
17.     send_mail(subject, message, from_email, recipient_list)
19. @receiver(user_registered)
20. def create_user_profile(sender, user, request, **kwargs):
21.     """
22.     创建用户档案
23.     """
24.     from .models import UserProfile
25.     UserProfile.objects.create(user=user)

复制代码

6、最佳实践

1. # myapp/signals.py - 定义所有信号
2. from django.dispatch import Signal
4. user_registered = Signal(providing_args=["user", "request"])
5. order_created = Signal(providing_args=["order", "user"])
7. # myapp/handlers.py - 定义所有信号处理器
8. from django.dispatch import receiver
9. from .signals import user_registered, order_created
11. @receiver(user_registered)
12. def handle_user_registered(sender, user, request, **kwargs):
13.     pass
15. @receiver(order_created)
16. def handle_order_created(sender, order, user, **kwargs):
17.     pass
19. # myapp/__init__.py - 导入信号处理器以确保它们被注册
20. default_app_config = 'myapp.apps.MyAppConfig'
22. # myapp/apps.py - 在应用配置中导入信号
23. from django.apps import AppConfig
25. class MyAppConfig(AppConfig):
26.     name = 'myapp'
27.    
28.     def ready(self):
29.         # 导入信号处理器
30.         import myapp.handlers

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！