如何使用SQLAlchemy对现有表执行高效批量插入/更新操作

核心方案说明 #

你当前使用的SQLAlchemy Core语法本身就是性能最优的选择，完全跳过了ORM层的实例创建、状态跟踪开销，和原生SQL执行效率差距极小，比bulk_insert_mappings、bulk_save_objects等ORM批量方法性能高15%-50%左右，完全符合你仅用字典/元组列表操作的需求。

批量插入优化 #

你现有写法可以稍作调整提升稳定性和性能，超大数据集建议分批插入（每批1000-5000条，避免超出数据库最大数据包限制）：

# 现有表反射代码保持不变即可
meta_data = sqlalchemy.MetaData(engine)
MyTable = sqlalchemy.Table('MyTable', meta_data, autoload=True, autoload_with=engine)

# 批量插入优化写法，用事务统一提交
BATCH_SIZE = 3000
data = [{'key1': 'value1', 'key2': 'value2'}, {'key1': 'value3', 'key2': 'value4'}]

with engine.begin() as conn:
    for i in range(0, len(data), BATCH_SIZE):
        batch = data[i:i+BATCH_SIZE]
        conn.execute(MyTable.insert(), batch)

批量更新实现 #

场景1：按唯一匹配条件批量更新不同行 #

如果你的字典列表每个条目都包含匹配字段（如主键/唯一键）和待更新字段，可以用bindparam构造动态更新语句，直接传字典列表执行，这是批量更新性能最高的方式：

from sqlalchemy import bindparam

# 假设根据key1匹配，更新key2字段
update_stmt = MyTable.update().where(
    MyTable.c.key1 == bindparam('_key1')
).values(
    key2 = bindparam('_key2')
)

# 构造参数列表，key前缀加下划线是为了和表字段名区分避免冲突
update_data = [
    {'_key1': 'value1', '_key2': 'new_value1'},
    {'_key1': 'value3', '_key2': 'new_value2'}
]

with engine.begin() as conn:
    conn.execute(update_stmt, update_data)

场景2：存在则更新、不存在则插入（Upsert） #

不同数据库的Upsert语法有差异，直接用Core提供的方言方法实现即可：

MySQL 版本

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(MyTable).values(data)
on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
    key2=insert_stmt.inserted.key2
)

with engine.begin() as conn:
    conn.execute(on_duplicate_key_stmt)

PostgreSQL 版本

from sqlalchemy.dialects.postgresql import insert

insert_stmt = insert(MyTable).values(data)
on_conflict_stmt = insert_stmt.on_conflict_do_update(
    index_elements=['key1'],
    set_={'key2': insert_stmt.excluded.key2}
)

with engine.begin() as conn:
    conn.execute(on_conflict_stmt)

额外性能优化建议 #

• 大批量数据操作建议暂时关闭表的索引/约束校验，操作完成后再重建，能提升数倍性能
• 避免在循环中单次提交，统一用事务批量提交可以极大减少IO开销
• 不需要返回插入行ID的场景下，可以在连接参数中关闭echo等调试输出，进一步降低开销

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