如何在Django Rest API中实现数据库新增条目管理员通知功能

先安装基础依赖：
pip install django djangorestframework

一、核心模型定义 #

需要两类模型：一类是实际业务中需要监控新增动作的业务表，另一类是专门存储管理员通知的表。直接在对应app的models.py里写如下代码，业务模型可以替换成你自己的现有表：

from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

# 示例：需要监控新增动作的业务模型，可直接替换为你的现有业务表
class UserSubmitRecord(models.Model):
    title = models.CharField(max_length=200, verbose_name="提交标题")
    content = models.TextField(verbose_name="提交内容")
    submit_user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="submit_records")
    create_time = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = "用户提交记录"
        ordering = ["-create_time"]

# 管理员通知存储表
class AdminNotification(models.Model):
    NOTIFY_TYPE_CHOICES = (
        ("new_record", "新增业务条目"),
    )
    title = models.CharField(max_length=200, verbose_name="通知标题")
    content = models.TextField(verbose_name="通知内容")
    notify_type = models.CharField(max_length=50, choices=NOTIFY_TYPE_CHOICES, default="new_record")
    target_admin = models.ForeignKey(User, on_delete=models.CASCADE, related_name="admin_notifications", limit_choices_to={"is_staff": True})
    is_read = models.BooleanField(default=False, verbose_name="是否已读")
    related_record_id = models.IntegerField(verbose_name="关联业务条目ID", null=True, blank=True)
    create_time = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = "管理员通知"
        ordering = ["-create_time"]

写完执行python manage.py makemigrations和python manage.py migrate生成数据库表。

二、新增条目自动触发通知逻辑 #

用Django原生的post_save信号监听业务模型的新增动作，只有新建条目时触发，修改条目不触发。
首先在app目录下新建signals.py文件：

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from .models import UserSubmitRecord, AdminNotification

User = get_user_model()

@receiver(post_save, sender=UserSubmitRecord)
def gen_new_record_notification(sender, instance, created, **kwargs):
    # 非新建操作直接返回
    if not created:
        return
    # 查询所有激活状态的管理员，可自定义筛选规则（比如按权限组、通知订阅配置过滤）
    admin_users = User.objects.filter(is_staff=True, is_active=True)
    # 批量生成通知，避免循环单条插入影响性能
    notify_bulk_list = []
    for admin in admin_users:
        notify_bulk_list.append(AdminNotification(
            title="收到新的用户提交记录",
            content=f"用户{instance.submit_user.username}提交了新内容：{instance.title}",
            target_admin=admin,
            related_record_id=instance.id
        ))
    AdminNotification.objects.bulk_create(notify_bulk_list)

信号必须手动注册才能生效，修改对应app的apps.py文件，在ready方法中导入信号文件：

from django.apps import AppConfig

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = '替换为你的实际app名称'

    def ready(self):
        import 替换为你的实际app名称.signals

到这一步，基础的自动触发逻辑已经跑通：只要业务表新增一条数据，会自动给所有管理员生成一条未读通知记录，可以先手动新建一条业务数据，查通知表确认逻辑正常。

三、DRF管理员通知接口开发 #

所有管理员对通知的操作全部走DRF接口实现，包含通知列表查询、详情查询、标记已读、未读数统计能力。
首先在serializers.py写序列化器：

from rest_framework import serializers
from .models import AdminNotification, UserSubmitRecord

class AdminNotificationSerializer(serializers.ModelSerializer):
    # 可选：关联返回对应业务条目的基础信息，方便管理员直接跳转
    related_biz_info = serializers.SerializerMethodField()

    def get_related_biz_info(self, obj):
        if not obj.related_record_id:
            return None
        biz_record = UserSubmitRecord.objects.filter(id=obj.related_record_id).first()
        if not biz_record:
            return None
        return {
            "id": biz_record.id,
            "title": biz_record.title,
            "submit_user": biz_record.submit_user.username,
            "create_time": biz_record.create_time.strftime("%Y-%m-%d %H:%M:%S")
        }

    class Meta:
        model = AdminNotification
        fields = ["id", "title", "content", "notify_type", "is_read", "related_biz_info", "create_time"]
        # 通知内容不允许管理员修改，全部设为只读
        read_only_fields = fields

然后在views.py写视图逻辑，强制管理员权限校验，只返回当前登录管理员自己的通知：

from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAdminUser
from .models import AdminNotification
from .serializers import AdminNotificationSerializer

class AdminNotificationViewSet(viewsets.ReadOnlyModelViewSet):
    serializer_class = AdminNotificationSerializer
    permission_classes = [IsAdminUser]

    def get_queryset(self):
        # 强制过滤：仅返回当前登录管理员的通知，避免越权查看其他管理员的通知
        qs = AdminNotification.objects.filter(target_admin=self.request.user)
        # 支持通过is_read参数筛选已读/未读通知
        is_read_param = self.request.query_params.get("is_read")
        if is_read_param is not None:
            qs = qs.filter(is_read=is_read_param.lower() == "true")
        return qs

    # 标记单条通知为已读
    @action(detail=True, methods=["POST"])
    def mark_read(self, request, pk=None):
        notify_obj = self.get_object()
        notify_obj.is_read = True
        notify_obj.save()
        return Response({"code": 0, "msg": "标记成功"}, status=status.HTTP_200_OK)

    # 标记当前管理员所有通知为已读
    @action(detail=False, methods=["POST"])
    def mark_all_read(self, request):
        self.get_queryset().filter(is_read=False).update(is_read=True)
        return Response({"code": 0, "msg": "全部标记已读成功"}, status=status.HTTP_200_OK)

    # 获取当前管理员未读通知总数，用于前端角标展示
    @action(detail=False, methods=["GET"])
    def unread_count(self, request):
        count = self.get_queryset().filter(is_read=False).count()
        return Response({"code": 0, "unread_count": count}, status=status.HTTP_200_OK)

最后在urls.py注册路由：

from rest_framework.routers import DefaultRouter
from . import views

router = DefaultRouter()
router.register(r"admin/notifications", views.AdminNotificationViewSet, basename="admin-notify")

urlpatterns = [
    # 其他业务路由
] + router.urls

完成后可以直接通过DRF的接口文档调试所有通知相关接口。

四、可选扩展 #

• 实时推送：如果需要管理员无刷新接收新通知，接入Django Channels实现WebSocket推送，在信号生成通知的逻辑里加一步往对应管理员的频道组推送消息即可，不需要改动现有核心逻辑。
• 多渠道触达：如果需要同步发邮件、企业微信/钉钉提醒，直接在信号的通知生成逻辑里加对应发送代码即可；如果第三方接口耗时较长，把发送任务丢到Celery异步队列执行，避免阻塞业务接口。
• 通知订阅配置：给管理员增加通知偏好设置，信号里查询管理员列表时过滤掉关闭了对应通知类型的账号即可。

踩坑提示 #

• 信号必须在apps.py的ready方法里导入，否则不会生效，不要漏这一步。
• 通知接口必须加IsAdminUser权限校验，同时查询时必须过滤target_admin=request.user，避免出现越权漏洞。
• 不要在信号逻辑里写耗时操作，高并发场景下会拖慢业务接口的响应速度，耗时操作一律走异步队列。

内容的提问来源于stack exchange，提问作者Зуфар Тангрибергенов