Django Rest Framework批量创建对象时，如何在保留模型唯一约束的同时返回重复字段并插入非重复数据？

嗨，我之前也碰到过一模一样的需求，批量插入时既要守住模型的唯一约束，又得区分处理重复和非重复数据，还要给前端返回清晰的结果。咱们一步步来解决这个问题：

首先得点出你现有代码的问题：当前逻辑只要有一条数据验证失败（比如重复的external_id），整个批量请求就会被打回，不会插入任何数据，这完全不符合你“插入非重复数据+返回重复ID”的核心需求。

核心思路 #

我们需要把传入的批量数据拆成三类分别处理：

1. 重复数据：external_id已经存在于数据库的条目
2. 有效非重复数据：可以正常插入的条目
3. 验证失败数据：除了重复之外，其他验证不通过的条目（比如phone_number长度不符合要求）

然后分别处理这三类数据，最后返回包含成功数据、重复错误、验证错误的组合响应。

修改后的代码实现 #

咱们直接修改views.py的post方法，这样逻辑更清晰，同时保留模型的unique约束（数据库层面依然会保证唯一性，能避免并发场景下的漏网之鱼）：

from django.db import IntegrityError
from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import Lead
from .serializers import LeadSerializer
from rest_framework.authentication import TokenAuthentication
from rest_framework.permissions import IsAuthenticated
from drf_yasg.utils import extend_schema

class LeadView(APIView):
    authentication_classes = [TokenAuthentication]
    permission_classes = [IsAuthenticated]

    @extend_schema(description="Insert campaign data", request=LeadSerializer(many=True), responses=None, tags=["Leads"])
    def post(self, request):
        incoming_data = request.data
        # 先校验传入格式是否为列表
        if not isinstance(incoming_data, list):
            return Response(
                {"status": "error", "data": "请求数据必须是Lead对象的列表"},
                status=status.HTTP_400_BAD_REQUEST
            )
        
        # 1. 拆分重复与非重复数据
        incoming_external_ids = [item.get("external_id") for item in incoming_data if item.get("external_id")]
        # 查询数据库中已存在的external_id
        existing_ids = set(Lead.objects.filter(
            external_id__in=incoming_external_ids
        ).values_list("external_id", flat=True))
        
        non_duplicate_data = []
        duplicate_errors = []
        validation_errors = []
        for item in incoming_data:
            ext_id = item.get("external_id")
            if not ext_id:
                # 缺少external_id直接归为验证错误
                validation_errors.append({"data": item, "errors": {"external_id": "该字段为必填项"}})
                continue
            if ext_id in existing_ids:
                duplicate_errors.append({
                    "external_id": ext_id,
                    "error": f"external_id {ext_id} 已存在"
                })
            else:
                non_duplicate_data.append(item)
        
        # 2. 处理非重复数据的验证与保存
        success_data = []
        if non_duplicate_data:
            serializer = LeadSerializer(data=non_duplicate_data, many=True)
            if serializer.is_valid():
                try:
                    # 尝试保存，捕获并发场景下的重复插入错误
                    serializer.save()
                    success_data = serializer.data
                except IntegrityError:
                    # 并发时可能出现：代码检查时ID不存在，但插入时被其他请求抢先创建
                    # 重新查询当前已存在的ID
                    current_existing = set(Lead.objects.filter(
                        external_id__in=[item["external_id"] for item in non_duplicate_data]
                    ).values_list("external_id", flat=True))
                    # 找出这次插入中新产生的重复ID
                    newly_duplicated = current_existing - existing_ids
                    # 分离成功保存和新重复的数据
                    success_items = [item for item in non_duplicate_data if item["external_id"] not in newly_duplicated]
                    success_data = LeadSerializer(Lead.objects.filter(external_id__in=[i["external_id"] for i in success_items]), many=True).data
                    # 新增重复错误
                    for ext_id in newly_duplicated:
                        duplicate_errors.append({
                            "external_id": ext_id,
                            "error": f"external_id {ext_id} 在插入过程中被其他请求抢先创建"
                        })
            else:
                # 收集每个条目单独的验证错误
                for idx, item_errors in enumerate(serializer.errors):
                    validation_errors.append({
                        "data": non_duplicate_data[idx],
                        "errors": item_errors
                    })
        
        # 3. 构造最终响应
        response_payload = {
            "status": "partial_success" if success_data or duplicate_errors or validation_errors else "success",
            "success_data": success_data,
            "duplicate_errors": duplicate_errors,
            "validation_errors": validation_errors
        }
        
        # 根据情况返回合适的状态码
        if success_data and (duplicate_errors or validation_errors):
            # 部分成功，用207多状态码（REST标准中表示批量部分成功）
            return Response(response_payload, status=status.HTTP_207_MULTI_STATUS)
        elif duplicate_errors or validation_errors:
            # 全失败
            return Response(response_payload, status=status.HTTP_400_BAD_REQUEST)
        else:
            # 全成功
            return Response(response_payload, status=status.HTTP_200_OK)

代码说明 #

1. 数据拆分：先提取所有传入的external_id，和数据库中已有的对比，把数据分成重复、非重复、缺失必填项三类。
2. 并发处理：用try-except捕获IntegrityError，处理并发场景下的“检查时不存在，插入时已存在”的问题，这是模型唯一约束带来的数据库层面保护，必须处理。
3. 响应构造：明确返回成功插入的数据、重复错误信息、其他验证错误，前端可以清晰地处理每部分结果。
4. 状态码选择：用207 MULTI_STATUS表示部分成功，这是REST API中处理批量部分成功的标准状态码。

为什么不修改serializer.is_valid()？ #

其实也可以在序列化器里重写is_valid，但批量数据的拆分和响应构造放在视图层更直观——序列化器的职责偏向于数据验证和模型转换，视图层负责请求处理和响应返回，这样代码职责更清晰，后期维护也更方便。

这样修改后，你就能完美实现需求：插入所有非重复的Lead，同时在响应中返回所有重复的external_id，而且模型的unique约束依然在数据库层面生效，不会出现数据不一致的问题。

备注：内容来源于stack exchange，提问作者Alejandro