Django REST项目API同步仓库数据至MySQL遇400错误排查

核心错误点分析 #

1. 模型拼写错误：Warehouse模型中on_delete=models.СASCAD使用了西里尔字母C，应为models.CASCADE，会导致数据库操作失败。
2. Serializer字段不匹配：
   • 定义了模型不存在的is_default_warehouse字段，触发序列化验证失败。
   • store字段错误定义为CharField，实际是ForeignKey，无法正确关联Store实例，导致验证不通过。
3. Create方法逻辑缺陷：
   • 遍历多个Store时，data变量被最后一个Store的数据覆盖，丢失之前的Store仓库信息。
   • 发现一个已存在仓库就直接返回，中断同步流程；创建第一个仓库后立即返回，未处理剩余仓库。
   • 创建Warehouse时未关联对应Store，导致store字段为null，违反业务逻辑且引发后续查询问题。
4. Update方法逻辑错误：
   • 遍历Warehouse时，处理第一个仓库后立即返回，未完成所有仓库更新。
   • 笔误：instance.name = wh['is_rfbs']应改为instance.is_rfbs = wh['is_rfbs']，错误将布尔值赋值给name字段。
   • save方法调用错误：self.instance.save缺少括号，instance.save(['name'])写法错误，正确应为instance.save(update_fields=['name'])。
5. Serializer Save方法错误：WarehouseUpdateSerializer中returself.instance是拼写错误，应为return self.instance。

修复后的代码示例 #

warehouses/models.py #

class Warehouse(models.Model):
    store = models.ForeignKey(
        Store,
        on_delete=models.CASCADE, null=True)  # 修正拼写错误
    warehouse_id = models.BigIntegerField(
        unique=True,
        null=True)
    name = models.CharField(
        max_length=150)
    is_rfbs = models.BooleanField(default=False)

serializers.py #

from rest_framework import serializers
from warehouses.models import Warehouse
from store.models import Store

class WarehouseSerializer(serializers.ModelSerializer):
    store = serializers.PrimaryKeyRelatedField(queryset=Store.objects.all())  # 使用正确的关联字段

    class Meta:
        model = Warehouse
        fields = ['store', 'warehouse_id', 'name', 'is_rfbs']  # 移除不存在的字段

class WarehouseUpdateSerializer(serializers.ModelSerializer):
    class Meta:
        model = Warehouse
        fields = ('name', 'is_rfbs',)

    def save(self, **kwargs):
        self.instance.name = self.validated_data["name"]
        self.instance.is_rfbs = self.validated_data["is_rfbs"]
        self.instance.save()  # 修正调用方式
        return self.instance  # 修正拼写错误

views.py #

from rest_framework import status
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet
from warehouses.models import Warehouse
from store.models import Store
from warehouses.serializers import WarehouseSerializer, WarehouseUpdateSerializer
from your_parser_module import Market  # 替换为实际Market类导入路径

class WarehouseApi(ModelViewSet):
    queryset = Warehouse.objects.all()
    serializer_class = WarehouseSerializer

    def get_queryset(self):
        return Warehouse.objects.filter(
            store__user_id=self.request.user.pk)\
            .order_by('-warehouse_id')

    def get_serializer_class(self):
        if self.request.method in ("PUT", "PATCH"):
            return WarehouseUpdateSerializer
        return WarehouseSerializer

    def create(self, request, *args, **kwargs):
        stores = Store.objects.filter(user=self.request.user, type=1)
        if not stores.exists():
            return Response(status=400, data={"Error": "No stores found for user with type=1"})

        for store in stores:
            api = Market(api_key=store.api_key, client_id=store.client_id)
            data = api.get_warehouses_list()
            
            if not data.get('result'):
                continue
            
            for wh in data['result']:
                warehouse_id = wh.get('warehouse_id')
                if not warehouse_id:
                    continue
                
                # 自动创建或更新，关联当前store
                Warehouse.objects.update_or_create(
                    warehouse_id=warehouse_id,
                    defaults={
                        'store': store,
                        'name': wh.get('name'),
                        'is_rfbs': wh.get('is_rfbs', False)
                    }
                )
        
        # 返回所有关联店铺的仓库
        warehouses = Warehouse.objects.filter(store__in=stores).order_by('-warehouse_id')
        if not warehouses.exists():
            return Response(status=400, data={"Error": "No warehouses synced"})
        
        serializer = WarehouseSerializer(warehouses, many=True)
        return Response(status=200, data=serializer.data)

    def update(self, request, *args, **kwargs):
        partial = kwargs.pop('partial', False)
        instance = self.get_object()
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        
        api = Market(api_key=instance.store.api_key, client_id=instance.store.client_id)
        data = api.get_warehouses_list()
        
        if not data.get('result'):
            return Response({'message': 'Store has no warehouses'}, status=status.HTTP_400_BAD_REQUEST)
        
        # 匹配远程对应仓库
        remote_warehouse = next((wh for wh in data['result'] if wh.get('warehouse_id') == instance.warehouse_id), None)
        if not remote_warehouse:
            return Response({'message': 'Warehouse not found in remote API'}, status=status.HTTP_404_NOT_FOUND)
        
        # 更新本地数据为远程最新值
        instance.name = remote_warehouse.get('name', instance.name)
        instance.is_rfbs = remote_warehouse.get('is_rfbs', instance.is_rfbs)
        instance.save()
        
        return Response({
            'data': WarehouseUpdateSerializer(instance).data,
            'status': status.HTTP_200_OK,
            'message': 'Warehouse updated successfully'
        })

关键修复说明 #

• 使用update_or_create替代手动判断存在性，简化逻辑并避免竞态条件。
• 修正Serializer字段类型，确保与模型匹配，消除验证失败根源。
• 调整Create方法逻辑，遍历所有Store并处理全部仓库，不中途返回。
• Update方法改为仅更新当前实例对应的远程仓库数据，逻辑更清晰高效。
• 修复所有拼写和语法错误，确保代码可正常运行。

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