如何创建可为Django模型生成信号的自定义装饰器？

需求描述 #

希望给Django模型添加自定义装饰器，自动生成带有固定业务逻辑的信号（仅关联模型不同），实现模型创建时向前端发送实时通知。

给定基础模型：

class Post(models.Model):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    title = models.CharField(max_length=100)

理想使用方式：

@custom_signal_decorator()
class Post(models.Model):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    title = models.CharField(max_length=100)

需要自动生成的信号逻辑（当模型实例创建时，向对应公司的WebSocket组发送通知）：

@receiver(post_save, sender=sender)
def notify_created(sender, instance, created, **kwargs):
    channel_layer = channels.layers.get_channel_layer()
    group_name = f'notifications_{instance.company.id}'
    async_to_sync(channel_layer.group_send)(
        group_name,
        {
            "type": "notify_change",
        }
    )

实现方案 #

1. 编写自定义装饰器 #

装饰器核心是接收模型类作为参数，动态创建信号处理函数并绑定到该模型的post_save信号上：

from django.db.models.signals import post_save
from django.dispatch import receiver
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

def custom_signal_decorator():
    def decorator(model_class):
        # 动态定义信号处理函数
        @receiver(post_save, sender=model_class)
        def notify_created(sender, instance, created, **kwargs):
            # 仅在实例创建时执行逻辑
            if not created:
                return
            channel_layer = get_channel_layer()
            group_name = f'notifications_{instance.company.id}'
            async_to_sync(channel_layer.group_send)(
                group_name,
                {
                    "type": "notify_change",
                }
            )
        # 返回原模型类，不改变模型结构
        return model_class
    return decorator

2. 在模型上使用装饰器 #

直接按照需求中的方式给模型添加装饰器即可：

@custom_signal_decorator()
class Post(models.Model):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    title = models.CharField(max_length=100)

3. 关键说明 #

• 装饰器采用嵌套结构：外层函数custom_signal_decorator()允许后续扩展参数（比如指定触发信号类型、自定义通知内容），内层decorator函数接收模型类并绑定信号。
• 信号处理函数中加入if not created: return判断，确保仅在实例创建时触发通知，符合“新帖子创建时发送通知”的需求。
• 装饰器仅在模型加载时动态注册信号，不修改模型原有结构，对模型无侵入。

替代方案（若装饰器方式不适用） #

如果觉得装饰器不够直观，可以创建抽象基类模型，将信号逻辑封装到基类中，需要实时通知的模型继承该基类：

class NotificationBaseModel(models.Model):
    class Meta:
        abstract = True

# 在signals.py中注册基类的信号（利用Django的信号继承特性）
@receiver(post_save, sender=NotificationBaseModel)
def notify_created(sender, instance, created, **kwargs):
    if not created or not hasattr(instance, 'company'):
        return
    channel_layer = get_channel_layer()
    group_name = f'notifications_{instance.company.id}'
    async_to_sync(channel_layer.group_send)(
        group_name,
        {
            "type": "notify_change",
        }
    )

# 使用时继承基类
class Post(NotificationBaseModel):
    company = models.ForeignKey(Company, on_delete=models.CASCADE)
    title = models.CharField(max_length=100)

内容的提问来源于stack exchange，提问作者4bs3nt