Django bulk_create批量创建时for循环查询外键实例性能低下如何优化

核心思路 #

你之前的写法哪怕提前调用了.all()，每次调用.get()还是会触发数据库查询，本质是把QuerySet当成了直接访问数据库的接口，没有用到内存缓存。正确做法是先批量查询所有需要用到的外键实例，转换成ID为键、实例为值的字典，后续循环直接在内存中查找，全程只需要4次外键表查询，没有循环内的数据库IO。

另外补充一个Django特性：外键字段支持直接通过字段名_id赋值ID值，不需要传入实例，如果你不需要用到外键实例的其他属性，直接赋值ID也可以省掉实例查询的步骤。

具体实现步骤 #

1. 先收集所有明细中用到的外键ID #

遍历一次明细列表，把所有需要用到的recinto、producto、estandar_producto、cotizacion的ID分别收集到集合里，自动去重避免无效查询。

2. 批量查询外键实例并转成字典 #

用__in查询一次性取出所有需要的实例，转成ID映射字典，全程每个外键表只查1次。

3. 循环构建明细实例时直接从字典取值 #

内存查找效率远高于数据库查询，哪怕有上万条明细也不会有性能问题。

优化后代码 #

class DetalleFichaAllViewSet(viewsets.ModelViewSet):
    serializer_class = DetalleFichaUpdateAllSerializer

    def create(self, request, *args, **kwargs):
        user = self.request.user
        data = request.data
        try:
            ficha = Ficha.objects.get(autor=user.id)
            DetalleFicha.objects.filter(ficha=ficha.id).delete()
        except Http404:
            pass

        # 更新Ficha信息
        now = datetime.now()
        date_time = now.strftime("%Y-%m-%d %H:%M")
        Ficha.objects.filter(autor=user.id).update(
            fecha_creacion=date_time, 
            autor=user, 
            nombre=data["nombreFicha"], 
            descripcion=data["descripcionFicha"]
        )
        ficha = Ficha.objects.filter(autor=user.id).last()

        # -------------------------- 优化部分开始 --------------------------
        # 第一步：收集所有用到的外键ID，用集合自动去重
        recinto_ids = set()
        producto_ids = set()
        estandar_producto_ids = set()
        cotizacion_ids = set()
        detalles = data["detalles"]
        for detalle in detalles:
            recinto_ids.add(detalle[1])
            producto_ids.add(detalle[10])
            estandar_producto_ids.add(detalle[9])
            if detalle[4]:  # 跳过空ID避免无效查询
                cotizacion_ids.add(detalle[4])
        
        # 第二步：批量查询所有外键实例，转成ID-实例字典，每个表仅查询1次
        recinto_map = {r.id: r for r in Recinto.objects.filter(id__in=recinto_ids)}
        producto_map = {p.id: p for p in Producto.objects.filter(id__in=producto_ids)}
        estandar_producto_map = {ep.id: ep for ep in EstandarProducto.objects.filter(id__in=estandar_producto_ids)}
        cotizacion_map = {c.id: c for c in Cotizacion.objects.filter(id__in=cotizacion_ids)}

        # 第三步：循环构建明细，全部走内存查询
        detalles_ficha = []
        for detalle in detalles:
            recinto = recinto_map[detalle[1]]
            producto = producto_map[detalle[10]]
            estandar_producto = estandar_producto_map[detalle[9]]
            cotizacion = cotizacion_map.get(detalle[4])  # 不存在自动返回None，无需额外捕获异常
            detalle_obj = DetalleFicha(
                carreras=detalle[0], 
                recinto=recinto, 
                nombre=detalle[2], 
                cantidad_a_comprar=detalle[3], 
                cotizacion=cotizacion, 
                valor_unitario=detalle[5], 
                valor_total=detalle[6], 
                documento=detalle[7], 
                fecha_cotizacion=detalle[8], 
                estandar_producto=estandar_producto, 
                producto=producto, 
                ficha=ficha
            )
            detalles_ficha.append(detalle_obj)
        # -------------------------- 优化部分结束 --------------------------
        
        DetalleFicha.objects.bulk_create(detalles_ficha)
        return Response(status=status.HTTP_200_OK)

额外优化点 #

• 如果不需要用到外键实例的任何属性，完全可以直接给外键的_id字段赋值，连实例查询都可以省掉，写法如下：

  detalle_obj = DetalleFicha(
      # 其他字段保持不变
      recinto_id=detalle[1],
      producto_id=detalle[10],
      estandar_producto_id=detalle[9],
      cotizacion_id=detalle[4] if detalle[4] in cotizacion_ids else None,
      ficha_id=ficha.id
  )
  

• 原代码中的调试打印语句建议上线前删除，避免影响性能和产生冗余日志。
• 如果明细量特别大（超过1万条），可以给bulk_create加batch_size参数，比如bulk_create(detalles_ficha, batch_size=1000)，避免单次插入SQL过长。

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