如何在SQLAlchemy中实现多行批量Upsert（Insert on Conflict）？

你可以直接结合SQLAlchemy的批量插入语法与on_conflict_do_update()实现单次请求的批量Upsert，无需提前查询区分插入/更新行，数据库会自动处理冲突逻辑。

核心实现方案 #

利用PostgreSQL的ON CONFLICT语法（SQLAlchemy通过on_conflict_do_update()封装），批量提交数据时，若主键/唯一索引冲突则执行更新，否则执行插入。

具体代码示例 #

假设你有一个ORM模型MyTable（对应数据库表my_table），主键为id，要批量Upsert多行数据：

1. 准备批量数据：

bulk_data = [
    {"id": 1, "data": "待插入数据1"},
    {"id": 2, "data": "待插入数据2"},
    {"id": 3, "data": "已存在id的更新数据"}  # 假设id=3在表中已存在
]

2. 构造批量Upsert语句：

from sqlalchemy.dialects.postgresql import insert
from sqlalchemy import create_engine
from models import MyTable  # 导入你的ORM模型

engine = create_engine("postgresql://user:password@host/dbname")
with engine.connect() as conn:
    # 构造批量插入语句
    insert_stmt = insert(MyTable).values(bulk_data)
    
    # 设置冲突处理：当id冲突时，更新data字段为插入时的新值
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["id"],  # 指定冲突判断的唯一索引/主键
        set_={"data": insert_stmt.excluded.data}  # 使用插入语句中待插入的data值更新现有行
    )
    
    # 执行语句并提交
    conn.execute(upsert_stmt)
    conn.commit()

关键说明 #

• insert_stmt.excluded：指代冲突发生时，原本要插入的那条数据的字段值，用它来更新现有行，能保证更新逻辑与插入数据一致。
• 无需预先查询数据库：数据库会自动检测每行数据是否存在冲突，无需分三次请求（查询→插入→更新），大幅提升效率。
• 灵活更新逻辑：set_参数支持更复杂的表达式，比如拼接字符串、使用函数等，例如：

  from sqlalchemy import func
  set_={"data": func.concat(insert_stmt.excluded.data, "（已更新）")}
  

与bulk_insert_mappings的关联 #

bulk_insert_mappings本质是批量插入的快捷方式，而上述方案直接复用了批量插入的语法，并通过on_conflict_do_update()扩展为Upsert，实现逻辑更统一，且无需额外拆分操作。

内容的提问来源于stack exchange，提问作者Sanjin Juric Fot