Django信号(Signal)机制的原理与应用，涵盖模型信号、自定义信号、信号注册的最佳实践

# 示例1：最简单的内置信号使用（技术栈：Django 4.2）
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import UserProfile

@receiver(post_save, sender=UserProfile)
def send_welcome_email(sender, instance, created, **kwargs):
    """新用户创建时发送欢迎邮件"""
    if created:
        subject = f"欢迎加入我们，{instance.username}！"
        message = "感谢注册我们的服务..."
        send_mail(subject, message, 'noreply@example.com', [instance.email])

# 示例2：自定义信号发送者
from django.dispatch import Signal

# 声明自定义信号
order_completed = Signal(providing_args=["order", "user"])

class Order:
    def complete_payment(self, user):
        """支付完成时触发信号"""
        self.status = 'PAID'
        self.save()
        order_completed.send(sender=self.__class__, order=self, user=user)

# 示例3：多种接收器注册方式
# 方式1：装饰器语法
@receiver(order_completed)
def update_inventory(sender, **kwargs):
    """订单完成时更新库存"""
    order = kwargs['order']
    for item in order.items.all():
        item.product.stock -= item.quantity
        item.product.save()

# 方式2：手动连接
def log_order_completion(sender, **kwargs):
    """记录订单完成日志"""
    logger.info(f"订单{kwargs['order'].id}已完成")
    
order_completed.connect(log_order_completion)

# 示例4：自动生成文章摘要
from django.db.models.signals import pre_save
from django.utils.text import Truncator

@receiver(pre_save, sender=BlogPost)
def generate_summary(sender, instance, **kwargs):
    """在保存前自动生成摘要"""
    if not instance.summary:
        instance.summary = Truncator(instance.content).chars(200)

# 示例5：结合Celery执行异步任务
from celery import shared_task

@receiver(post_save, sender=Comment)
def notify_subscribers(sender, instance, created, **kwargs):
    """新评论创建时异步通知订阅用户"""
    if created:
        send_comment_notification.delay(comment_id=instance.id)

@shared_task
def send_comment_notification(comment_id):
    comment = Comment.objects.get(id=comment_id)
    # 执行耗时通知逻辑...

# 示例6：操作日志追踪
from django.contrib.admin.models import LogEntry, CHANGE

@receiver(post_save, sender=Product)
def log_price_change(sender, instance, **kwargs):
    """记录价格变更历史"""
    if instance.tracker.has_changed('price'):
        old_price = instance.tracker.previous('price')
        LogEntry.objects.log_action(
            user_id=instance.last_modified_by.id,
            content_type=ContentType.objects.get_for_model(instance),
            object_id=instance.id,
            change_message=f"价格从 {old_price} 调整为 {instance.price}"
        )

# 示例7：推荐的信号注册方式
# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    def ready(self):
        # 确保信号处理程序只注册一次
        from . import signals  # noqa

# 示例8：使用dispatch_uid防止重复注册
post_save.connect(
    update_search_index,
    sender=Article,
    dispatch_uid="update_search_index_for_article"
)

# 示例9：测试信号处理程序
from django.test import TestCase
from django.db.models.signals import post_save

class SignalTestCase(TestCase):
    def test_welcome_email_sent(self):
        """测试新用户注册触发邮件发送"""
        with self.captureOnCommitCallbacks() as callbacks:
            User.objects.create(username='test', email='test@example.com')
            
        self.assertEqual(len(callbacks), 1)
        self.assertIn('send_welcome_email', str(callbacks[0]))

# 替代方案示例：重写save方法
class Order(models.Model):
    def save(self, *args, **kwargs):
        created = not self.pk
        super().save(*args, **kwargs)
        if created:
            self.send_confirmation_email()

# 错误示例：在post_save中保存对象导致循环
@receiver(post_save, sender=Product)
def update_rating(sender, instance, **kwargs):
    instance.update_rating()  # 会导致新的保存操作！

# 正确做法：使用update_fields或条件判断
@receiver(post_save, sender=Product)
def update_rating(sender, instance, **kwargs):
    if not kwargs.get('update_fields'):
        instance.update_rating(update_fields=['rating'])