Django中带自定义主键的M2M字段模型创建操作问题

问题背景 #

我定义了两个Django模型：Vehicle使用plate作为自定义主键，Center与Vehicle通过多对多字段关联。在单元测试中，创建Center时传入已存在的Vehicle数据，触发了plate的唯一性错误。尝试在CenterView的perform_create方法中拦截流程但没成功，想知道有没有对应的验证步骤可以重写，或者用类似get_or_create的逻辑来处理这种场景，寻求拦截验证步骤的可行方案。

模型代码 #

from django.db import models
from django.conf import settings

class Vehicle(models.Model):
    """Vehicle object."""
    plate = models.CharField(max_length=255, primary_key=True)
    brand = models.CharField(max_length=255)
    model = models.CharField(max_length=255)
    owner = models.CharField(max_length=255)
    user = models.ForeignKey(
            settings.AUTH_USER_MODEL,
            on_delete=models.CASCADE,
    )

class Center(models.Model):
    """Center object."""
    user = models.ForeignKey(
            settings.AUTH_USER_MODEL,
            on_delete=models.CASCADE,
    )
    name = models.CharField(max_length=255)
    vehicles = models.ManyToManyField('Vehicle', related_name='vehicle_plate')

测试代码 #

def test_create_center_with_existing_vehicles(self):
    """Test creating a center with existing vehicle."""
    vehicle_1 = Vehicle.objects.create(user=self.user, plate='ab123cd', brand='brand', model='model', owner='bla bla')

    payload = {
            'name': 'first center',
            'vehicles': [{'plate': 'ab123cd', 'brand': 'brand', 'model': 'model', 'owner': 'bla bla'}]
    }

    res = self.client.post(CENTERS_URL, payload, format='json')

错误信息 #

{'vehicles': [{'plate': [ErrorDetail(string='vehicle with this plate already exists.', code='unique')]}, {}]}

核心原因 #

默认的DRF序列化器处理嵌套多对多数据时，会直接尝试创建新的Vehicle对象，不会自动检查对象是否已存在，因此当传入已存在的plate（主键）时，会触发数据库的唯一性约束错误。

解决方案 #

方案1：重写序列化器的create方法（推荐） #

在CenterSerializer中手动处理vehicles字段，对每个车辆数据使用get_or_create逻辑，实现“存在则关联，不存在则创建”的效果：

from rest_framework import serializers
from .models import Center, Vehicle

class VehicleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Vehicle
        fields = ['plate', 'brand', 'model', 'owner']  # 移除user字段，避免前端传入，由后端自动赋值

class CenterSerializer(serializers.ModelSerializer):
    vehicles = VehicleSerializer(many=True)

    class Meta:
        model = Center
        fields = ['id', 'name', 'vehicles']  # user字段由视图自动赋值

    def create(self, validated_data):
        # 提取车辆数据并从验证数据中移除
        vehicles_data = validated_data.pop('vehicles')
        # 获取当前请求用户，确保车辆归属正确
        current_user = self.context['request'].user
        # 创建Center对象，关联当前用户
        center = Center.objects.create(user=current_user, **validated_data)
        
        # 批量处理车辆关联
        for vehicle_data in vehicles_data:
            # 使用plate作为唯一标识，存在则获取，不存在则创建
            vehicle, _ = Vehicle.objects.get_or_create(
                plate=vehicle_data['plate'],
                defaults={**vehicle_data, 'user': current_user}
            )
            center.vehicles.add(vehicle)
        return center

方案2：自定义验证逻辑提前拦截 #

如果不想完全重写create方法，可以在序列化器中添加validate_vehicles方法，提前检查车辆是否存在，将已存在的对象直接加入验证结果，避免后续创建：

class CenterSerializer(serializers.ModelSerializer):
    vehicles = VehicleSerializer(many=True)

    class Meta:
        model = Center
        fields = ['id', 'name', 'vehicles']

    def validate_vehicles(self, value):
        validated_items = []
        current_user = self.context['request'].user
        for vehicle_data in value:
            plate = vehicle_data.get('plate')
            if not plate:
                raise serializers.ValidationError('plate字段不能为空')
            # 检查当前用户下是否已存在该车辆
            try:
                vehicle = Vehicle.objects.get(plate=plate, user=current_user)
                validated_items.append(vehicle)
            except Vehicle.DoesNotExist:
                # 不存在则保留原始数据，后续创建
                validated_items.append(vehicle_data)
        return validated_items

    def create(self, validated_data):
        vehicles = validated_data.pop('vehicles')
        current_user = self.context['request'].user
        center = Center.objects.create(user=current_user, **validated_data)
        
        for item in vehicles:
            if isinstance(item, Vehicle):
                # 已存在的车辆直接关联
                center.vehicles.add(item)
            else:
                # 新车辆创建后关联
                vehicle = Vehicle.objects.create(user=current_user, **item)
                center.vehicles.add(vehicle)
        return center

视图层配合（可选） #

如果你的CenterView是基于CreateAPIView或ModelViewSet，确保序列化器能获取到请求上下文：

from rest_framework import generics
from .models import Center
from .serializers import CenterSerializer

class CenterCreateView(generics.CreateAPIView):
    queryset = Center.objects.all()
    serializer_class = CenterSerializer

    # 手动设置user的方式（可选）
    def perform_create(self, serializer):
        serializer.save(user=self.request.user)

测试验证 #

修改序列化器后，运行原测试代码，res应该返回201 Created状态码，且Center对象会正确关联已存在的Vehicle，不会触发唯一性错误。

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