如何在Django的bulk_create操作中捕获批量插入的失败行？

核心思路 #

bulk_create默认是原子操作，只要有一行数据验证失败或触发数据库约束冲突，整个批次都会回滚。所以核心方案是先逐行验证数据合法性，把合法实例和错误行分开，再对合法数据执行批量保存，同时收集错误行信息反馈给用户做选择。

具体实现步骤 #

1. 解析文件并逐行验证 #

无论是CSV还是Excel，先逐行读取数据，用Django模型的clean()方法或自定义逻辑做验证，拆分合法/非法数据：

import csv
from django.core.exceptions import ValidationError
from .models import YourTargetModel

def parse_and_validate_file(file):
    valid_instances = []
    invalid_rows = []
    
    # CSV处理示例（Excel可改用openpyxl库按类似逻辑读取）
    reader = csv.DictReader(file)
    # 行号从2开始，因为第一行是表头
    for row_num, row_data in enumerate(reader, start=2):
        try:
            # 实例化模型对象
            instance = YourTargetModel(
                field_a=row_data['field_a'],
                field_b=row_data['field_b'],
                # 映射其他字段...
            )
            # 触发模型内置验证（含字段类型、自定义clean逻辑）
            instance.clean()
            valid_instances.append(instance)
        except ValidationError as e:
            # 收集验证错误：行号、原始数据、错误详情
            invalid_rows.append({
                'row_number': row_num,
                'raw_data': row_data,
                'errors': e.message_dict
            })
        except Exception as e:
            # 捕获其他意外错误（比如类型转换失败）
            invalid_rows.append({
                'row_number': row_num,
                'raw_data': row_data,
                'errors': str(e)
            })
    
    return valid_instances, invalid_rows

2. 处理验证结果，对接用户选择 #

拿到拆分后的结果后，根据用户操作执行对应逻辑：

from django.http import JsonResponse

def upload_handler(request):
    if request.method == 'POST':
        upload_file = request.FILES.get('data_file')
        if not upload_file:
            return JsonResponse({'error': '请选择要上传的文件'}, status=400)
        
        valid_instances, invalid_rows = parse_and_validate_file(upload_file)
        
        if invalid_rows:
            # 返回错误信息，让用户选择跳过错误行或重新上传
            return JsonResponse({
                'has_errors': True,
                'invalid_rows': invalid_rows,
                'valid_count': len(valid_instances),
                'invalid_count': len(invalid_rows)
            })
        else:
            # 无错误则直接批量保存
            YourTargetModel.objects.bulk_create(valid_instances)
            return JsonResponse({'success': True, 'saved_count': len(valid_instances)})

3. 大型文件优化 #

如果文件行数极多（比如几十万行），逐行验证可能导致内存占用过高，可以改成分批验证：

def parse_and_validate_large_file(file, batch_size=1000):
    valid_instances = []
    invalid_rows = []
    reader = csv.DictReader(file)
    current_batch = []
    
    for row_num, row_data in enumerate(reader, start=2):
        current_batch.append((row_num, row_data))
        if len(current_batch) >= batch_size:
            # 处理当前批次
            for rn, rd in current_batch:
                try:
                    instance = YourTargetModel(**rd)
                    instance.clean()
                    valid_instances.append(instance)
                except Exception as e:
                    invalid_rows.append({
                        'row_number': rn,
                        'raw_data': rd,
                        'errors': str(e)
                    })
            current_batch = []
    
    # 处理剩余未达批次的行
    if current_batch:
        for rn, rd in current_batch:
            try:
                instance = YourTargetModel(**rd)
                instance.clean()
                valid_instances.append(instance)
            except Exception as e:
                invalid_rows.append({
                    'row_number': rn,
                    'raw_data': rd,
                    'errors': str(e)
                })
    
    return valid_instances, invalid_rows

4. 关键注意点 #

• 数据库约束兜底：即使通过了模型clean()验证，仍可能触发数据库层面的约束（如唯一键冲突）。如果要忽略这类冲突，可以用bulk_create(ignore_conflicts=True)（仅支持PostgreSQL、MySQL 8.0+等），但需谨慎使用避免数据丢失。
• 事务保障：如果需要保证合法行的原子性保存，可包裹事务：

  from django.db import transaction
  
  with transaction.atomic():
      YourTargetModel.objects.bulk_create(valid_instances)
  

• 错误信息可读性：返回给用户的错误要明确标注行号和具体字段问题，方便快速修正。

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