使用Django REST Framework上传文件：新增正常但无法更新同名数据

在使用Django REST Framework实现文件上传导入Category数据时，新数据能正常创建，但遇到CSV中name字段（已转小写）与数据库中现有条目重复时，无法更新现有条目，反而会触发唯一约束报错。

原代码 #

class CategoryUploadView(APIView):
    authentication_classes = [OAuth2Authentication]
    permission_classes = [AllowAny]    
    parser_classes = [MultiPartParser]

    def post(self, request):
        file_obj = request.FILES['file']
        decoded_file = file_obj.read().decode('utf-8').splitlines()
        reader = csv.DictReader(decoded_file)
        for row in reader:
            serializer = CategorySerializer(data=row)
            if serializer.is_valid():
                serializer.save()
            else:
                return Response(serializer.errors, status=400)
        return Response({"message": "Data imported successfully"}, status=200)

class CategorySerializer(serializers.ModelSerializer):
    class Meta:
        model=Category
        fields=['id', 'name']

    def to_internal_value(self, data):
        data = data.copy()
        data['name'] = data.get('name', None).lower()
        return super().to_internal_value(data)

class Category(models.Model):
    name = models.CharField(max_length=32, unique=True)
    description = models.CharField(max_length=255)
    update_on = models.DateField(auto_now=True)

    def __str__(self) -> str:
        return self.name 

核心原因 #

当前代码每次调用serializer.save()都会尝试创建新的Category实例，而name字段是唯一约束，当CSV中存在同名（转小写后）条目时，会直接触发数据库唯一约束错误，且没有逻辑去匹配现有条目执行更新操作。

修复方案 #

我们可以在循环处理CSV每行数据时，先根据小写后的name查询现有记录，存在则更新，不存在则创建。同时优化错误处理，避免遇到单个错误就终止整个导入流程。

方案一：直接在视图层处理（高效直观） #

修改CategoryUploadView的post方法：

from django.db import IntegrityError

class CategoryUploadView(APIView):
    authentication_classes = [OAuth2Authentication]
    permission_classes = [AllowAny]    
    parser_classes = [MultiPartParser]

    def post(self, request):
        file_obj = request.FILES['file']
        decoded_file = file_obj.read().decode('utf-8').splitlines()
        reader = csv.DictReader(decoded_file)
        errors = []
        for idx, row in enumerate(reader, 1):
            # 先处理name转小写，和serializer逻辑保持一致
            row_copy = row.copy()
            row_copy['name'] = row_copy.get('name', '').lower()
            try:
                # 查询现有条目，不存在则创建
                category, created = Category.objects.get_or_create(
                    name=row_copy['name'],
                    defaults={**row_copy}
                )
                # 如果不是新建条目，执行更新
                if not created:
                    # 仅更新模型中存在的字段
                    for key, value in row_copy.items():
                        if key != 'name' and hasattr(category, key):
                            setattr(category, key, value)
                    category.save()
            except IntegrityError as e:
                errors.append(f"第{idx}行数据出错: {str(e)}")
            except Exception as e:
                errors.append(f"第{idx}行数据处理失败: {str(e)}")
        
        if errors:
            return Response({"errors": errors}, status=400)
        return Response({"message": "Data imported successfully"}, status=200)

方案二：复用Serializer验证逻辑（符合DRF规范） #

如果希望通过Serializer处理验证和更新逻辑，可以在初始化Serializer时传入现有实例：

class CategoryUploadView(APIView):
    authentication_classes = [OAuth2Authentication]
    permission_classes = [AllowAny]    
    parser_classes = [MultiPartParser]

    def post(self, request):
        file_obj = request.FILES['file']
        decoded_file = file_obj.read().decode('utf-8').splitlines()
        reader = csv.DictReader(decoded_file)
        errors = []
        for idx, row in enumerate(reader, 1):
            try:
                # 先处理name转小写
                row_copy = row.copy()
                row_copy['name'] = row_copy.get('name', '').lower()
                # 查询现有条目
                category = Category.objects.get(name=row_copy['name'])
                # 传入实例执行更新
                serializer = CategorySerializer(instance=category, data=row_copy)
            except Category.DoesNotExist:
                # 无实例则执行创建
                serializer = CategorySerializer(data=row_copy)
            
            if serializer.is_valid():
                serializer.save()
            else:
                errors.append(f"第{idx}行数据验证失败: {serializer.errors}")
        
        if errors:
            return Response({"errors": errors}, status=400)
        return Response({"message": "Data imported successfully"}, status=200)

额外优化建议 #

• 批量处理性能：如果导入数据量极大，可以收集待创建和待更新的条目，最后用bulk_create和bulk_update批量操作，减少数据库查询次数。
• 字段白名单：可以在视图中定义允许更新的字段列表，避免CSV中的无关字段导致意外更新。

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