sqlalchemy 如何写“存在则更新，不存在则插入”（upsert）

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
stmt = insert(User).values(
id=123,
name='Alice',
email='alice@example.com'
)
stmt = stmt.on_conflict_do_update(
indexelements=['id'],  # 冲突依据（主键或唯一索引列）
set=dict(name=stmt.excluded.name, email=stmt.excluded.email)  # 更新字段
)
conn.execute(stmt)

• MySQL：用 insert().on_duplicate_key_update()

from sqlalchemy.dialects.mysql import insert
stmt = insert(User).values(id=123, name='Alice', email='alice@example.com')
stmt = stmt.on_d
uplicate_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
stmt = insert(User).values(id=123, name='Alice', email='alice@example.com')
stmt = stmt.on_conflict_do_update(
indexelements=['id'],
set=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'},
]
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__，但这不推荐

# ai  # 后端  # app  # session  # select  # mysql 

相关栏目： <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 AI推广<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 SEO优化<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 技术百科<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 谷歌推广<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 百度推广<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 网络营销<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 案例网站<？ｍｕｍａ echo $count; ?> 】 <？ｍｕｍａ $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 精选文章<？ｍｕｍａ echo $count; ?> 】