Django中如何让信号接收器监听特定用户信号并推送审核通知

安全阿岳

2026-10-6

解决方案

你当前采用post_save信号实现的思路存在两个天然缺陷：

信号属于模型层逻辑，和请求生命周期完全解耦，无法直接获取执行审核操作的管理员身份、也拿不到对应的request对象
原生Django的messages框架绑定当前请求的用户，无法直接跨用户给博文作者发送消息，也无法实现无需刷新的跨标签页实时通知

以下是两种适配不同需求的实现方案：

方案一：无额外依赖的非实时实现（用户刷新页面即可看到通知）

不需要用到信号，把通知触发逻辑直接迁移到审核操作对应的视图层即可，如果你用Django自带的Admin做审核，直接重写BlogPost对应的ModelAdmin的save_model方法即可：

from django.contrib import admin
from django.contrib import messages
from django.contrib.sessions.models import Session
from django.utils import timezone
from .models import BlogPost

class BlogPostAdmin(admin.ModelAdmin):
    def save_model(self, request, obj, form, change):
        # 先获取修改前的审核状态
        original_approved = BlogPost.objects.get(pk=obj.pk).approved if change else False
        super().save_model(request, obj, form, change)
        # 仅当审核状态发生变化时触发通知
        if change and original_approved != obj.approved:
            author = obj.user
            # 遍历所有未过期的session，找到作者的所有登录会话
            valid_sessions = Session.objects.filter(expire_date__gte=timezone.now())
            for session in valid_sessions:
                session_data = session.get_decoded()
                if session_data.get('_auth_user_id') == str(author.id):
                    # 按照messages框架的存储格式写入作者的session
                    if '_messages' not in session_data:
                        session_data['_messages'] = []
                    if obj.approved:
                        msg_item = (messages.SUCCESS, '你的博文已审核通过', '')
                    else:
                        msg_item = (messages.ERROR, '你的博文未通过审核', '')
                    session_data['_messages'].append(msg_item)
                    # 保存修改后的session
                    session.session_data = Session.objects.encode(session_data)
                    session.save()

admin.site.register(BlogPost, BlogPostAdmin)

这种方案下，作者无论在哪个标签页，只要刷新页面就会自动加载session中存储的消息并显示，符合Django原生messages框架的使用逻辑，不需要额外安装依赖。

方案二：支持实时跨标签页推送的实现

如果需要做到审核通过后，作者已经打开的所有页面不需要刷新就能立刻弹出通知，需要使用WebSocket协议实现主动推送，Django生态中常用Channels库实现WebSocket功能，实现步骤如下：

安装配置Django Channels，配置Redis作为频道层存储（生产环境）
编写WebSocket消费者，每个登录用户连接后自动加入专属的通知频道组
审核状态变化时，向作者的专属频道组推送通知消息
所有页面引入前端JS代码，监听WebSocket消息并弹出提示

核心代码示例

消费者代码（consumers.py）

import json
from channels.generic.websocket import AsyncWebsocketConsumer

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user = self.scope["user"]
        if not self.user.is_authenticated:
            await self.close()
            return
        # 每个用户对应一个专属通知组
        self.group_name = f"user_{self.user.id}_notify"
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.group_name, self.channel_name)

    # 接收组推送的消息并发送给前端
    async def send_notify(self, event):
        await self.send(text_data=json.dumps({
            "level": event["level"],
            "content": event["content"]
        }))

审核视图中推送消息的逻辑

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

# 在审核状态变化的逻辑中加入以下代码
channel_layer = get_channel_layer()
author = obj.user
if obj.approved:
    level, content = "success", "你的博文已审核通过"
else:
    level, content = "error", "你的博文未通过审核"
# 向作者的专属组推送消息
async_to_sync(channel_layer.group_send)(
    f"user_{author.id}_notify",
    {"type": "send_notify", "level": level, "content": content}
)

前端公共JS代码（所有页面引入）

const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'
const notifySocket = new WebSocket(`${wsProtocol}//${window.location.host}/ws/notify/`)
notifySocket.onmessage = (e) => {
    const data = JSON.parse(e.data)
    // 可替换为项目中使用的提示组件逻辑，比如Toast、Alert等
    if (data.level === 'success') {
        alert(`成功：${data.content}`)
    } else {
        alert(`错误：${data.content}`)
    }
}

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