sqlalchemy实现upsert有三种方式：一是core层数据库原生语法（postgresql用on_conflict_do_update、mysql用on_duplicate_key_update、sqlite用on_conflict_do_update），高效且避免竞态；二是orm层bulk_upsert_mappings（2.0+），批量处理、不触发事件；三是merge()，自动查再更/插但有性能开销。
SQLAlchemy 提供了多种方式实现“存在则更新，不存在则插入”（upsert），具体写法取决于你用的是 Core 还是 ORM，以及后端数据库是否原生支持 upsert（如 PostgreSQL 的 ON CONFLICT、MySQL 的 ON DUPLICATE KEY UPDATE、SQLite 的 ON CONFLICT REPLACE/DO UPDATE）。
使用 Core：database-native upsert（推荐，高效）
这是最直接、性能最好的方式，利用数据库原生语法，避免先查再插/更的竞态问题和额外查询开销。
-
PostgreSQL：用
insert().on_conflict_do_update()
from sqlalchemy.dialects.postgresql import insert <p>stmt = insert(User).values( id=123, name='Alice', email='alice@example.com' ) stmt = stmt.on_conflict_do_update( index<em>elements=['id'], # 冲突依据（主键或唯一索引列） set</em>=dict(name=stmt.excluded.name, email=stmt.excluded.email) # 更新字段 ) conn.execute(stmt)
-
MySQL：用
insert().on_duplicate_key_update()
from sqlalchemy.dialects.mysql import insert <p>stmt = insert(User).values(id=123, name='Alice', email='alice@example.com') stmt = stmt.on_duplicate_key_update( name=stmt.inserted.name, email=stmt.inserted.email ) conn.execute(stmt)
-
SQLite（≥3.24.0）：用
insert().on_conflict_do_update()
from sqlalchemy.dialects.sqlite import insert <p>stmt = insert(User).values(id=123, name='Alice', email='alice@example.com') stmt = stmt.on_conflict_do_update( index<em>elements=['id'], set</em>=dict(name=stmt.excluded.name, email=stmt.excluded.email) ) conn.execute(stmt)
使用 ORM：bulk_upsert_mappings（批量高效，适合多条）
适用于一次性处理多条记录，底层调用对应 dialect 的原生 upsert，比逐条 session.merge() 快得多。
- 要求模型定义了主键或唯一约束（用于判断“存在”）
- 只支持 2.0+ 版本（
session.bulk_upsert_mappings()）
data = [
{'id': 123, 'name': 'Alice', 'email': 'a1@example.com'},
{'id': 456, 'name': 'Bob', 'email': 'b1@example.com'},
]
<p>session.bulk_upsert_mappings(User, data)
session.commit()
注意：bulk_upsert_mappings 不触发 ORM 事件（如 @event.listens_for），也不做对象实例化，纯数据层操作。
兼容性兜底：用 merge()（简单但有代价）
如果数据库不支持原生 upsert，或你需要 ORM 对象生命周期管理（如事件、关系加载），可用 session.merge()：
- 它会先按主键/identity 查询是否存在；存在则更新属性，不存在则新建
- 每次调用都产生一次 SELECT 查询（可能变慢，尤其高并发时）
- 适合单条、低频、开发/测试场景
user = User(id=123, name='Alice', email='alice@example.com') merged = session.merge(user) # 自动判断插入或更新 session.commit()
关键注意事项
- 确保目标列（如
id或email）上有主键或唯一约束，否则 upsert 无法判断冲突 - PostgreSQL 的
index_elements必须匹配某个唯一索引（可以是复合唯一索引的一部分） - MySQL 要求冲突列属于某个
PRIMARY KEY或UNIQUE KEY - ORM 的
merge()按主键比较，不支持按非主键字段（如 email）判断“存在”——除非你重写__hash__和__eq__，但这不推荐
