为何SQLAlchemy的session.merge与bulk_update_mappings远慢于手动全量查询更新？

• session.merge性能差的核心原因是单次单行查询开销：默认每处理一条数据，merge都会单独向数据库发送1次SELECT查询匹配主键记录是否存在，500条数据对应500次独立的数据库请求，搭配远程MySQL的网络往返开销累加，就会出现数秒的延迟。
• bulk_update_mappings性能差的核心原因是默认不做批量语句合并：在SQLAlchemy 1.4版本中，bulk_update_mappings默认会为每一条待更新数据生成独立的UPDATE语句，逐条发往数据库执行，同样会产生500次网络往返，远程数据库场景下开销被放大。如果没有配置PyMySQL的批量执行参数，这一问题会更明显。

现有方法2的合规性 #

方法2是完全合规的实现，适用场景为表总数据量较小（千级别以内）、无高并发修改冲突的业务场景。它的性能优势来自于仅产生2次网络往返：1次全表查询拉取所有数据，1次提交批量推送所有变更，所有更新逻辑在内存中完成，天然避免了多次请求的开销。
可以进一步优化方法2的匹配逻辑，将双重循环替换为字典映射，降低时间复杂度：

with Session() as session:
    db_rows = session.query(DemoClass).all()
    # 构建id到行对象的映射，O(1)匹配
    id_to_row = {row.id: row for row in db_rows}
    for item in data:
        db_row = id_to_row.get(item['id'])
        if db_row:
            db_row.col1 = item['col1']
            db_row.col2 = item['col2']
            db_row.col3 = item['col3']
            db_row.col4 = item['col4']
            db_row.col5 = item['col5']
            db_row.col6 = item['col6']
            db_row.local_modified = datetime.strptime(item['local_modified'], '%Y-%m-%d %H:%M:%S')
    session.commit()

更高性能的通用upsert方案 #

场景1：使用MySQL数据库（最优方案） #

直接用MySQL原生的ON DUPLICATE KEY UPDATE语法，通过SQLAlchemy Core实现，500条数据仅需1次数据库请求，性能远高于其他方案：

from sqlalchemy.dialects.mysql import insert

with Session() as session:
    insert_stmt = insert(DemoClass).values(data)
    # 定义主键冲突时的更新逻辑
    upsert_stmt = insert_stmt.on_duplicate_key_update(
        col1=insert_stmt.inserted.col1,
        col2=insert_stmt.inserted.col2,
        col3=insert_stmt.inserted.col3,
        col4=insert_stmt.inserted.col4,
        col5=insert_stmt.inserted.col5,
        col6=insert_stmt.inserted.col6,
        local_modified=insert_stmt.inserted.local_modified
    )
    session.execute(upsert_stmt)
    session.commit()

场景2：表数据量大、不适合全表查询 #

先批量查询所有待更新的id对应的记录，再在内存中做更新/新增，避免全表扫描也避免单行查询：

with Session() as session:
    # 提取所有待处理的id，批量查询
    target_ids = [item['id'] for item in data]
    exist_rows = session.query(DemoClass).filter(DemoClass.id.in_(target_ids)).all()
    id_to_row = {row.id: row for row in exist_rows}
    add_list = []
    for item in data:
        row = id_to_row.get(item['id'])
        if row:
            # 更新已有记录
            row.col1 = item['col1']
            row.col2 = item['col2']
            # 其他字段同理
            row.local_modified = datetime.strptime(item['local_modified'], '%Y-%m-%d %H:%M:%S')
        else:
            # 新增不存在的记录
            add_list.append(DemoClass(**item))
    if add_list:
        session.add_all(add_list)
    session.commit()

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