如何用SQLAlchemy实现支持数据库生成默认值的分批次批量插入

你当前实现性能较低的核心原因是：默认配置下SQLAlchemy的bulk_save_objects会尝试为每一条插入的记录获取数据库生成的主键/默认值，PostgreSQL原生executemany不支持批量返回多行生成值，会被拆分为单条INSERT语句执行，导致效率大幅下降。

以下是不同场景下的可行方案：

方案1：无需返回生成字段的最简优化 #

如果插入完成后不需要马上获取每条记录的id等数据库生成字段，直接在bulk_save_objects中关闭返回默认值的行为即可：

def bulk_create_messages_for_mymodel(self, objects_list):
    objects = bulk_create_objects(MyModel, objects_list)
    with self.session.begin():
        self.session.bulk_save_objects(objects, return_defaults=False)

该改法会将插入操作合并为真正的批量executemany调用，性能和Django默认bulk_create效果完全一致。

方案2：需要返回生成字段的高性能实现 #

如果需要拿到插入后的生成字段（如主键id用于后续逻辑），直接使用SQLAlchemy Core的INSERT语句配合PostgreSQL原生RETURNING特性实现批量返回：

from sqlalchemy import insert

BATCH_SIZE = 10000

def bulk_create_with_returning(model, instance_list):
    returned_instances = []
    # 按批次拆分避免单条SQL过长
    for idx in range(0, len(instance_list), BATCH_SIZE):
        batch = instance_list[idx:idx+BATCH_SIZE]
        # 构造批量插入+全字段返回语句
        insert_stmt = insert(model).values(batch).returning(model)
        batch_res = self.session.scalars(insert_stmt).all()
        returned_instances.extend(batch_res)
    self.session.commit()
    return returned_instances

该方案单次批量插入仅执行1条SQL，同时返回所有批次记录的生成字段，性能远高于默认bulk_save_objects带返回值的实现。

方案3：SQLAlchemy 2.0 版本最优配置 #

如果你使用的是SQLAlchemy 2.0及以上版本，直接调整引擎参数开启原生批量插入优化即可，无需修改业务代码：

engine = create_engine(
    settings.MY_DATABASE,
    # 开启PostgreSQL专用批量插入合并特性
    use_insertmanyvalues=True,
    # 单批次插入的记录数，和Django batch_size参数等效
    insertmanyvalues_page_size=10000
)

开启后SQLAlchemy会自动将所有批量插入操作合并为INSERT INTO table (col1, col2) VALUES (v1, v2), (v3, v4)...的格式，同时原生支持批量返回生成字段，性能为所有方案中最高。

注意事项 #

• 批次大小建议设置在5000~20000区间，过大可能触发PostgreSQL单条SQL长度限制、事务锁数量限制，过小会降低批量操作收益
• 如果表存在较多索引、触发器，可在批量插入前临时禁用，插入完成后重建，可进一步提升30%~50%的插入速度
• 不要手动为id这类数据库生成字段赋值，避免PostgreSQL序列值和实际表数据不一致

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