Django REST Framework批量POST时校验记录存在并跳过重复项的方案

核心思路 #

校验与数据处理逻辑优先放在序列化层实现，符合DRF分层设计规范，复用性更高，视图层仅按需做响应逻辑的定制。

方案1：兼容所有Django版本，支持返回跳过明细 #

步骤1：自定义批量序列化器处理去重逻辑 #

# serializers.py
from rest_framework import serializers
from .models import Data

class DataListSerializer(serializers.ListSerializer):
    def create(self, validated_data):
        # 先对提交数据本身去重，避免列表内重复数据导致冲突
        unique_items = {}
        for item in validated_data:
            union_key = (item["key_1"], item["key_2"])
            unique_items[union_key] = item
        cleaned_data = list(unique_items.values())

        # 查询数据库中已存在的联合键
        exist_keys = set(Data.objects.values_list("key_1", "key_2"))

        # 过滤掉已存在的记录，仅保留新增数据
        new_items = [
            item for item in cleaned_data
            if (item["key_1"], item["key_2"]) not in exist_keys
        ]

        # 批量插入提升性能
        return Data.objects.bulk_create([Data(**item) for item in new_items])


class DataSerializer(serializers.ModelSerializer):
    class Meta:
        model = Data
        fields = "__all__"
        # 指定批量创建时使用的序列化器
        list_serializer_class = DataListSerializer

步骤2：视图层按需定制（可选） #

原有create方法无需修改即可正常运行，如果需要在响应中明确告知用户跳过的数量，可以小幅调整：

# views.py
from rest_framework import viewsets, status
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from .models import Data
from .serializers import DataSerializer

class DataViewSet(viewsets.ModelViewSet):
    queryset = Data.objects.all()
    serializer_class = DataSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_fields = ['key_1', 'key_2']

    def create(self, request, *args, **kwargs):
        is_batch = isinstance(request.data, list)
        serializer = self.get_serializer(data=request.data, many=is_batch)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)

        if is_batch:
            return Response({
                "total_received": len(request.data),
                "success_created": len(serializer.data),
                "skipped_duplicate": len(request.data) - len(serializer.data),
                "data": serializer.data
            }, status=status.HTTP_201_CREATED, headers=headers)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

方案2：Django 4.1+ 最优性能方案 #

如果使用Django 4.1及以上版本，无需提前查询数据库，可以直接利用数据库的唯一约束特性，通过bulk_create的ignore_conflicts参数自动跳过重复记录，代码更简洁、性能更高：

# serializers.py 中的DataListSerializer可简化为
class DataListSerializer(serializers.ListSerializer):
    def create(self, validated_data):
        return Data.objects.bulk_create(
            [Data(**item) for item in validated_data],
            ignore_conflicts=True
        )

该方案的唯一缺点是无法准确获取被跳过的记录明细，如果不需要返回跳过信息，这是批量场景下的最优选择。

常见问题解答 #

1. 是否必须重写ModelViewSet的create方法？
   不是必须。仅当需要自定义批量创建的响应内容时才需要调整，否则原有默认create方法完全可以配合自定义序列化器实现需求。
2. 为什么不把去重逻辑写在视图层？
   序列化层是DRF规定的校验、数据处理层，逻辑放在这里可以在所有用到该序列化器的场景复用，避免视图层臃肿，符合单一职责设计原则。

注意事项 #

• 发起POST请求时建议直接传Python列表给requests.post的json参数，无需手动序列化，requests会自动设置正确的Content-Type: application/json请求头。
• Django 4.2+版本推荐使用UniqueConstraint代替unique_together定义联合唯一约束，写法更灵活：

# models.py
class Data(models.Model):
    key_1 = models.CharField(max_length=64, blank=True, null=True)
    key_2 = models.CharField(max_length=64, blank=True, null=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["key_1", "key_2"], name="unique_key_pair")
        ]

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