如何通过SQLAlchemy ORM Session批量执行Upsert操作？

针对多个对象执行基于主键的Upsert，你可以通过以下两种方式实现：

1. 基础循环方案 #

这是最直接的实现方式，对每个目标对象依次调用session.merge()：

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
with Session() as session:
    # 假设companies是包含多个company对象的列表
    for company in companies:
        session.merge(company)
    session.commit()

该方案逻辑简单，完全复用merge的原生逻辑，适合对象数量不多的场景。

2. 批量优化方案（SQLAlchemy 2.0+） #

如果处理的对象数量较大，循环merge会产生多次数据库查询，性能较低。此时可以结合数据库原生Upsert语法（如MySQL的ON DUPLICATE KEY UPDATE、PostgreSQL的ON CONFLICT），配合ORM的批量插入API提升效率：

示例（以MySQL为例） #

from sqlalchemy import insert
from sqlalchemy.orm import sessionmaker
from your_module import Company  # 导入你的ORM模型

Session = sessionmaker(engine)
with Session() as session:
    # 将对象转换为字典列表
    company_dicts = [
        {"id": comp.id, "name": comp.name, "address": comp.address}
        for comp in companies
    ]
    
    # 构造批量Upsert语句
    stmt = insert(Company).values(company_dicts).on_duplicate_key_update(
        name=inserted.name,
        address=inserted.address
        # 按需列出需要更新的字段
    )
    
    session.execute(stmt)
    session.commit()

这种方式仅生成单条批量Upsert SQL，性能远高于循环merge，但需要注意：

• 依赖数据库原生Upsert语法，不同数据库的写法存在差异
• 需要手动指定更新字段，无法自动同步对象的所有属性

关于merge的补充说明 #

Session.merge()会检查源实例的主键属性，尝试与会话中相同主键的实例进行协调。如果本地未找到，则尝试根据主键从数据库加载对象；若仍未找到，则创建新实例。随后将源实例的每个属性状态复制到目标实例，方法返回目标实例；原始源实例保持未修改状态，若未关联会话则不会关联。

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