SQLAlchemy 2.0异步ORM深度嵌套一对多批量插入性能优化咨询

针对A→B→C层级嵌套数据的批量插入性能问题，这里提供几个实用的实现方式：

1. 用bulk_save_objects批量保存关联ORM实例 #

相比逐个调用session.add()，bulk_save_objects能大幅减少SQL执行次数，同时保留ORM的关联关系处理能力。只需提前构建好层级关联的ORM实例，一次性批量提交即可。

示例代码（假设ORM模型为ModelA、ModelB、ModelC）：

from sqlalchemy.ext.asyncio import AsyncSession

async def bulk_insert_nested(session: AsyncSession, json_data_list: list):
    a_instances = []
    for a_json in json_data_list:
        # 构建C层实例
        c_instances = [ModelC(**c_item) for c_item in a_json["b_list"][0]["c_list"]]
        # 构建B层实例并关联C
        b_instance = ModelB(**a_json["b_list"][0], c_list=c_instances)
        # 构建A层实例并关联B
        a_instance = ModelA(**a_json, b_list=[b_instance])
        a_instances.append(a_instance)
    
    # 批量保存所有层级实例
    session.bulk_save_objects(a_instances)
    await session.commit()

注：如果依赖before_insert这类模型事件钩子，需要添加return_defaults=True参数（会有少量性能开销）。

2. 分阶段批量插入+手动维护外键 #

追求极致性能的话，可以按表层级分阶段插入，手动处理外键关联：

• 先批量插入C表数据，获取插入后的主键
• 把C的主键映射到B的外键字段，批量插入B表并获取B的主键
• 最后把B的主键映射到A的外键字段，批量插入A表

示例代码：

from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncSession

async def staged_bulk_insert(session: AsyncSession, json_data_list: list):
    # 提取各层级原始数据并记录关联映射
    all_c_data = []
    all_b_data = []
    all_a_data = []
    b_to_a_idx = {}
    c_to_b_idx = {}

    for a_idx, a_json in enumerate(json_data_list):
        # 提取A表数据（剔除嵌套的B数据）
        a_data = {k: v for k, v in a_json.items() if k != "b_list"}
        all_a_data.append(a_data)
        
        for b_idx, b_json in enumerate(a_json["b_list"]):
            # 提取B表数据（剔除嵌套的C数据）
            b_data = {k: v for k, v in b_json.items() if k != "c_list"}
            b_key = (a_idx, b_idx)
            b_to_a_idx[b_key] = a_idx
            all_b_data.append(b_data)
            
            for c_idx, c_json in enumerate(b_json["c_list"]):
                c_key = (a_idx, b_idx, c_idx)
                c_to_b_idx[c_key] = (a_idx, b_idx)
                all_c_data.append(c_json)
    
    # 批量插入C表，返回主键
    c_result = await session.execute(insert(ModelC).values(all_c_data).returning(ModelC.id))
    c_ids = [row[0] for row in c_result.all()]
    # 给B数据绑定C的外键
    for c_seq, c_key in enumerate(c_to_b_idx.keys()):
        a_idx, b_idx = c_to_b_idx[c_key]
        all_b_data[b_idx]["c_id"] = c_ids[c_seq]  # 替换为实际外键字段名
    
    # 批量插入B表，返回主键
    b_result = await session.execute(insert(ModelB).values(all_b_data).returning(ModelB.id))
    b_ids = [row[0] for row in b_result.all()]
    # 给A数据绑定B的外键
    for b_seq, b_key in enumerate(b_to_a_idx.keys()):
        a_idx = b_to_a_idx[b_key]
        all_a_data[a_idx]["b_id"] = b_ids[b_seq]  # 替换为实际外键字段名
    
    # 批量插入A表
    await session.execute(insert(ModelA).values(all_a_data))
    await session.commit()

这种方式性能最优，但需要手动处理数据映射，适合超大规模数据插入场景。

3. 用run_sync调用同步ORM的批量关联插入 #

SQLAlchemy同步ORM的批量关联支持更成熟，可以通过AsyncSession.run_sync在异步环境中复用同步逻辑，兼顾关联便利性和批量性能：

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session

async def sync_style_bulk_insert(session: AsyncSession, json_data_list: list):
    def sync_insert_logic(sync_session: Session):
        a_instances = []
        for a_json in json_data_list:
            c_instances = [ModelC(**c_item) for c_item in a_json["b_list"][0]["c_list"]]
            b_instance = ModelB(**a_json["b_list"][0], c_list=c_instances)
            a_instance = ModelA(**a_json, b_list=[b_instance])
            a_instances.append(a_instance)
        sync_session.bulk_save_objects(a_instances)
    
    await session.run_sync(sync_insert_logic)
    await session.commit()

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