PostgreSQL上使用on_conflict_do_update实现Upsert

本篇文章主要是结合我之前面试的各种经历和实战开发中遇到的问题解决经验整理的，希望这篇《SQLAlchemy 实现“存在则更新，不存在则插入”（Upsert）通常可以通过 merge() 方法或者结合 on_conflict_do_update（在 PostgreSQL 中）来实现。以下是几种常见的方法：✅ 方法一：使用 session.merge()这是 SQLAlchemy 提供的原生方式，适用于大多数数据库。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from your_model import YourModel engine = create_engine('sqlite:///your.db') Session = sessionmaker(bind=engine) session = Session() # 假设你有一个对象 obj = YourModel(id=1, name='Test') # merge 会根据主键判断是插入还是更新 session.merge(obj) session.commit()说明：如果 id 已经存在，merge() 会执行更新。如果 id 不存在，会执行插入。需要确保模型中定义了主键（id）。✅ 方法二：PostgreSQL 使用 on_conflict_do_update如果你使用的是 PostgreSQL，可以使用 on_conflict_do_update 来实现 Upsert。安装依赖（如果没安装）：pip install psycopg2-binary示例代码： from sqlalchemy import Column, Integer, String, create_engine from sqlalchemy.dialects.postgresql import》对你有很大帮助！欢迎收藏，分享给更多的需要的朋友学习~

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)
</p>

• 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)
</p>

• 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)
</p>

使用 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()
</p>

注意：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__，但这不推荐

今天关于《PostgreSQL上使用on_conflict_do_update实现Upsert》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！