Django如何限制特定用户关联对象创建数量 等效GORM钩子的实现方法

Django 提供了多种和 GORM 钩子逻辑对齐的实现方式，可根据业务场景选择，以下是常用的3种方案：

1. 重写模型 save() 方法（最接近 GORM 钩子的实现，全局生效） #

该方案和 GORM 的 BeforeSave 钩子逻辑完全一致，在任务写入数据库前自动触发校验，所有调用 Task.save() 的场景都会生效，避免重复写校验逻辑。

from django.db import models
from django.core.exceptions import ValidationError
from django.contrib.auth import get_user_model

User = get_user_model()

class Task(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="tasks")
    title = models.CharField(max_length=200)
    content = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    # 配置单用户最大任务数
    MAX_TASK_PER_USER = 50

    def save(self, *args, **kwargs):
        # 仅新增任务时校验，更新现有任务不触发
        if not self.pk:
            # count() 直接执行数据库 COUNT 查询，效率极高，不会拉取全量数据
            user_task_count = Task.objects.filter(user=self.user).count()
            if user_task_count >= self.MAX_TASK_PER_USER:
                raise ValidationError(f"单个用户最多只能创建{self.MAX_TASK_PER_USER}条任务")
        super().save(*args, **kwargs)

如果需要返回 JSON 格式的错误提示，在视图层捕获异常即可，以 DRF 视图为例：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import Task
from .serializers import TaskSerializer

class TaskCreateView(APIView):
    def post(self, request):
        serializer = TaskSerializer(data=request.data, context={"request": request})
        if serializer.is_valid():
            try:
                serializer.save(user=request.user)
                return Response(serializer.data, status=status.HTTP_201_CREATED)
            except ValidationError as e:
                return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

如果需要避免并发请求穿透校验（同一时间多个创建请求同时查询到任务数为49，都通过校验写入），可以加行锁和事务保障数据一致性：

from django.db import transaction

def save(self, *args, **kwargs):
    if not self.pk:
        with transaction.atomic():
            # 锁住对应用户行，避免其他事务同时修改该用户的任务数量
            User.objects.select_for_update().get(id=self.user.id)
            user_task_count = Task.objects.filter(user=self.user).count()
            if user_task_count >= self.MAX_TASK_PER_USER:
                raise ValidationError(f"单个用户最多只能创建{self.MAX_TASK_PER_USER}条任务")
    super().save(*args, **kwargs)

该加锁方案仅锁定当前用户的单行数据，不会影响其他用户的任务创建操作，性能损耗极低

2. 信号（Signal）实现（适合解耦场景） #

如果不想把校验逻辑耦合在模型的 save 方法中，可以用 Django 内置的 pre_save 信号实现，逻辑和 GORM 钩子完全一致：

from django.db.models.signals import pre_save
from django.dispatch import receiver
from .models import Task

@receiver(pre_save, sender=Task)
def check_task_limit(sender, instance, **kwargs):
    if not instance.pk:
        user_task_count = Task.objects.filter(user=instance.user).count()
        if user_task_count >= Task.MAX_TASK_PER_USER:
            raise ValidationError(f"单个用户最多只能创建{Task.MAX_TASK_PER_USER}条任务")

该方案适合多个模型需要统一管理同类校验逻辑的场景，缺点是逻辑分散，排查问题成本略高

3. 序列化器校验（仅接口层生效，适合轻量场景） #

如果仅通过前端接口创建任务，没有其他后台逻辑直接创建 Task 实例，也可以把校验逻辑写在 DRF 序列化器中：

from rest_framework import serializers
from .models import Task

class TaskSerializer(serializers.ModelSerializer):
    class Meta:
        model = Task
        fields = ["id", "title", "content", "created_at"]
    
    def validate(self, attrs):
        user = self.context["request"].user
        if Task.objects.filter(user=user).count() >= Task.MAX_TASK_PER_USER:
            raise serializers.ValidationError(f"单个用户最多只能创建{Task.MAX_TASK_PER_USER}条任务")
        return attrs

该方案不需要额外捕获异常，DRF 会自动返回 JSON 格式的错误提示，缺点是仅调用序列化器创建时生效，覆盖范围有限

注意事项 #

• 所有方案中的 count() 查询均为数据库层面的聚合查询，即使单表数据量达到百万级别也不会有性能问题
• 高并发场景优先选择加行锁的重写 save 方案，一致性和性能平衡最好

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