Django导入CSV文件时for循环处理耗时过长如何优化

以下优化按性能提升幅度从高到低排序，修改后1000条记录处理耗时可降到秒级：

核心问题修复 #

1. 移除循环内重复数据库查询（最大耗时点） #

你当前代码每遍历一行CSV，就执行一次AsignacionContact.objects.get(id=self.object.pk)，1000行就会发起1000次完全重复的SQL查询，属于典型的N+1查询问题。
实际上self.object.pk是循环全程固定不变的值，且Django外键支持直接通过字段名_id传关联主键值，完全不需要查询关联模型实例，零数据库开销就能完成外键赋值。

错误写法：
idasignacion=AsignacionContact.objects.get(id=self.object.pk)
优化写法（循环外提前定义，全程复用）：
asignacion_contact_id = self.object.pk
实例化模型时直接传：idasignacion_id=asignacion_contact_id

2. 删除循环内无意义的调试打印 #

循环里的print('inicio de for')会在每一行处理时触发标准输出IO，1000行就打印1000次日志，IO开销远高于纯CPU计算，是第二大耗时来源，直接删除即可。

3. 提前做字段类型转换，减少ORM隐式开销 #

你的模型中corriente_no_vencida_actual、corrente_vencida、total_deuda_corriente、cuota_minima_agente是整数类型字段，但当前代码直接传入CSV读取到的字符串，Django ORM在实例化模型时会自动做类型转换、字段合法性校验，每一行都要跑一遍全字段校验逻辑，累积开销很高。
可以在循环外定义统一的单元格值处理函数，提前做去空格、空值处理、整数转换，减少ORM层的隐式计算。

4. 简化CSV读取逻辑 #

当前用codecs.iterdecode包装文件流再传给csv.reader，多了一层迭代解码的开销，直接以指定编码打开文件流传给csv.reader即可，减少不必要的迭代包装。

5. 调整批量插入批次大小 #

你当前设置的5000条一批插入阈值过大，主流关系型数据库单次bulk_create的最优批次是1000-2000条，批次过大会导致单次SQL语句过长，数据库解析开销上升，反而拖慢速度。

优化后完整代码参考 #

import csv

class AsignacionCreateView(CreateView):
    model=AsignacionContact
    form_class=AsignacionForm
    template_name='gestionAsignacion/asignacion_contact.html'
    success_url = reverse_lazy('gestionAsignacion:asignacion_contact')
    
    def form_valid(self, form):
        isvalid = super().form_valid(form)
        csv_file = self.request.FILES['file']
        # 提前固定外键值，全程无重复查询
        asignacion_contact_id = self.object.pk
        # 定义需要转整数的字段索引
        INT_FIELD_INDEX = {5, 6, 7, 8}
        # 循环外定义值处理函数，避免重复创建函数开销
        def parse_cell(idx, raw_val):
            val = raw_val.strip() if raw_val else None
            if not val:
                return None
            if idx in INT_FIELD_INDEX:
                return int(val)
            return val

        asignaciones = []
        # 简化CSV读取逻辑
        with csv_file.open('r', encoding='utf-8') as f:
            reader = csv.reader(f, delimiter=",")
            next(reader) # 跳过表头
            for row in reader:
                asignacion = AsignacionSur(
                    idasignacion_id=asignacion_contact_id,
                    producto=parse_cell(0, row[0]),
                    contrato=parse_cell(1, row[1]),
                    nombre_suscriptor=parse_cell(2, row[2]),
                    tipo_servicio=parse_cell(3, row[3]),
                    total_deuda=parse_cell(4, row[4]),
                    corriente_no_vencida_actual=parse_cell(5, row[5]),
                    corrente_vencida=parse_cell(6, row[6]),
                    total_deuda_corriente=parse_cell(7, row[7]),
                    cuota_minima_agente=parse_cell(8, row[8]),
                    politica_surtigas=parse_cell(9, row[9]),
                    categoria=parse_cell(10, row[10]),
                    estrato=parse_cell(11, row[11]),
                    ref_anio=parse_cell(12, row[12]),
                    historico_ref=parse_cell(12, row[12]),
                    ciclo=parse_cell(14, row[14]),
                    medidor=parse_cell(15, row[15]),
                    lectura=parse_cell(16, row[16]),
                    plan_acuerdo=parse_cell(17, row[17]),
                    descripcion_barrio=parse_cell(18, row[18]),
                    direccion=parse_cell(19, row[19]),
                    dias_deuda=parse_cell(20, row[20])
                )
                asignaciones.append(asignacion)
                # 调整批次大小为1000，适配数据库最优插入性能
                if len(asignaciones) >= 1000:                
                    AsignacionSur.objects.bulk_create(asignaciones)
                    asignaciones = []
            if asignaciones:
                AsignacionSur.objects.bulk_create(asignaciones)              
        return isvalid

额外优化建议 #

• 导入操作前临时关闭Django的DEBUG模式，避免框架记录所有SQL查询、请求上下文带来的额外内存和CPU开销。
• 如果CSV文件超过10万行，不要在web请求中同步处理，改成Celery、Django-Q等异步任务执行，避免请求超时，同时可以给用户展示导入进度。
• 你的AsignacionSur模型主键是自定义的IntegerField而非自增字段，bulk_create时要确保主键值不重复，避免插入报错。

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