Django signal system - event-driven programming model | Daoman PythonAI

#django signal system - event-driven programming model

📂 Stage: Part 2 - Advanced Features 🎯 Difficulty level: Intermediate ⏰ Estimated study time: 2 hours 🎒 Prerequisite knowledge: 模型设计与ORM操作

Core concept: What is a Django signal?

Django signal is the standard implementation of Observer Pattern, used for loosely coupled application component communication: when an "event" occurs (such as model save/delete, request start/end), the signal is triggered and all receivers who "subscribe" to the signal are automatically notified.

Three core roles of signals

RoleDescription
SignalAbstract carrier of events (such aspost_save）
SenderThe subject that triggers the signal (such as a specificUserModel class)
ReceiverFunction/class method that responds to the signal (such as "Send welcome email")

Get started quickly: the most basic signal usage

1. Introduce dependencies

from django.dispatch import Signal, receiver
from django.db.models.signals import post_save
from myapp.models import User

2. Receiver registration (two methods)

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    """新用户注册成功后发送欢迎邮件"""
    if created:
        # 实际项目请用Celery异步
        print(f"向 {instance.email} 发送欢迎邮件")

Method 2: Manual connection

def create_user_profile(sender, instance, created, **kwargs):
    """新用户注册成功后创建档案"""
    if created:
        from myapp.models import UserProfile
        UserProfile.objects.create(user=instance)

# 在 apps.py 的 ready() 中调用此代码
post_save.connect(create_user_profile, sender=User)

3. Signal triggering (no need to write additional code)

when executingUser.objects.create_user(...)oruser.save()hour,post_saveThe signal will trigger automatically.

Commonly used built-in signals

Django has three types of commonly used signals built-in, covering core scenarios.

1. Model signal (most commonly used)

Automatically triggered when operating the database model:

from django.db.models.signals import (
    pre_init,    # 模型实例化前（几乎不用）
    post_init,   # 模型实例化后
    pre_save,    # 模型保存前
    post_save,   # 模型保存后（★核心：区分created/更新）
    pre_delete,  # 模型删除前
    post_delete, # 模型删除后
    m2m_changed, # 多对多关系变更
)

Model signal example: automatically update article word count

from django.db.models.signals import pre_save
from django.dispatch import receiver
from myapp.models import Article

@receiver(pre_save, sender=Article)
def auto_count_words(sender, instance, **kwargs):
    """文章保存前自动统计字数"""
    if instance.content:
        instance.word_count = len(instance.content.strip().split())

2. Request signal

Triggered when processing an HTTP request:

from django.core.signals import (
    request_started,    # 请求开始
    request_finished,   # 请求结束
    got_request_exception, # 请求异常（可用于全局错误记录）
)

3. Application/Project Signal

from django.db.models.signals import (
    pre_migrate,  # 迁移执行前
    post_migrate, # 迁移执行后（★核心：创建初始数据）
)

Apply Signal Example: Create Super Administrator after Migration

from django.db.models.signals import post_migrate
from django.dispatch import receiver
from django.contrib.auth.models import User

@receiver(post_migrate)
def create_superuser(app_config, **kwargs):
    """仅在myapp迁移完成后创建超级管理员"""
    if app_config.name == 'myapp':
        if not User.objects.filter(username='admin').exists():
            User.objects.create_superuser(
                'admin', 'admin@example.com', 'admin123'
            )

Custom signals: meet personalized needs

When built-in signals are not enough, custom signals can be created.

1. Define custom signals

# myapp/signals.py
from django.dispatch import Signal

# 定义一个「订单完成」的信号，携带order、user参数
order_completed = Signal()

2. Send custom signals

# myapp/services.py
from myapp.models import Order
from myapp.signals import order_completed

def complete_order(order_id):
    """标记订单完成并发送信号"""
    order = Order.objects.get(id=order_id)
    order.status = 'COMPLETED'
    order.save()
    
    # 发送信号（★显式调用）
    order_completed.send(
        sender=Order,
        order=order,
        user=order.user
    )

3. Receive custom signals

# myapp/signals.py
from django.dispatch import receiver
from .models import Order

@receiver(order_completed, sender=Order)
def send_order_confirmation(sender, order, user, **kwargs):
    """订单完成后发送确认通知"""
    print(f"向 {user.email} 发送订单 {order.id} 的确认")

Best Practices and Pitfall Guidelines

⚠️ Pitfall 1: Signals must be imported when the application starts

If the signal module is not imported, the receiver will not take effect! must be inapps.pyofready()Explicit import in:

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        # 显式导入信号模块，触发装饰器的注册逻辑
        import myapp.signals

⚠️ Pitfall 2: The default is synchronous blocking

If the signal receiver performs time-consuming operations (such as sending emails, calling APIs), it will directly block the main thread! Asynchronous tasks must be used (such as Celery):

# myapp/signals.py
from celery import shared_task

@shared_task
def async_send_welcome_email(user_id):
    """异步发送欢迎邮件"""
    from django.contrib.auth.models import User
    user = User.objects.get(id=user_id)
    print(f"异步向 {user.email} 发送邮件")

@receiver(post_save, sender=User)
def trigger_async_email(sender, instance, created, **kwargs):
    """只触发异步任务，不阻塞"""
    if created:
        async_send_welcome_email.delay(instance.id)

⚠️ Pitfall 3: Avoid signal loop calls

if inpost_saveCalled again in the receiverinstance.save(), will trigger an infinite loop! There are two solutions:

  1. Add processing flag
@receiver(post_save, sender=Article)
def safe_article_update(sender, instance, created, **kwargs):
    if getattr(instance, '_in_signal', False):
        return
    instance._in_signal = True
    try:
        # 安全的更新逻辑
        if not instance.slug:
            instance.slug = f"article-{instance.id}"
            instance.save()
    finally:
        instance._in_signal = False
  1. Useupdate_fieldsorQuerySet.update()
@receiver(post_save, sender=Article)
def safe_article_update(sender, instance, created, **kwargs):
    if not instance.slug and created:
        # QuerySet.update() 不会触发 post_save
        Article.objects.filter(id=instance.id).update(slug=f"article-{instance.id}")

⚠️ Pitfall 4: Temporarily disconnect the signal during large data operations

bulk_createNo single signal will be triggered, but if you must operate row by row, remember to temporarily disconnect the signal to improve performance:

from contextlib import contextmanager
from django.db.models.signals import post_save

@contextmanager
def temp_disconnect_signal(signal, receiver, sender):
    """临时断开信号的上下文管理器"""
    signal.disconnect(receiver, sender=sender)
    try:
        yield
    finally:
        signal.connect(receiver, sender=sender)

# 使用示例
with temp_disconnect_signal(post_save, send_welcome_email, User):
    # 批量导入1000个测试用户，不会触发欢迎邮件
    for i in range(1000):
        User.objects.create_user(f"test{i}", f"test{i}@example.com", "123")

Summary of this chapter

Django signals are a powerful tool for decoupling application components, but they must be used with caution:

  1. ✅ Suitable for scenarios: data synchronization, cache invalidation, audit logs, notification push
  2. ❌ Not suitable for scenarios: core business logic (signals are "optional notifications", not "required dependencies")
  3. 🚀 Optimization suggestions: use asynchronous tasks, avoid loops, and disconnect signals during large-scale operations

🔗 Recommended related tutorials

🏷️ tag cloud:django信号 事件驱动 观察者模式 解耦 Celery 性能优化