SQLAlchemy 2.0+ 中 PostgreSQL 用 on_conflict_do_update(index_elements=[...]) 实现 upsert，需严格匹配唯一约束字段；SQLite 用 on_conflict_do_replace()；MySQL 需手动 prefix_with("ON DUPLICATE KEY UPDATE") 并查 ROW_COUNT()；返回完整记录须配合 returning() 且注意数据库版本与 ORM 刷新问题。
SQLAlchemy 2.0+ 的 insert().on_conflict_do_update() 怎么用
PostgreSQL 和 SQLite（3.24+）原生支持 upsert，SQLAlchemy 2.0+ 将其封装为 on_conflict_do_update()（PostgreSQL）和 on_conflict_do_replace()（SQLite）。关键不是“有没有”，而是“冲突键怎么写对”——漏掉 index_elements 或写错字段名，会直接报 IntegrityError 而非执行更新。
示例：用户表按 email 唯一约束做 upsert：
from sqlalchemy.dialects.postgresql import insert <p>stmt = insert(User).values( email="alice@example.com", name="Alice", updated_at=func.now() ) stmt = stmt.on_conflict_do_update( index<em>elements=["email"], # 必须匹配唯一索引/主键字段，不能写成 ["id"] 除非是主键冲突 set</em>=dict( name=stmt.excluded.name, updated_at=func.now() ) ) session.execute(stmt) session.commit()
-
index_elements不是表字段任意组合，必须严格对应数据库中已存在的唯一约束或主键定义 -
stmt.excluded是 PostgreSQL 特有命名，指代本次 INSERT 中被拒绝的那行数据；SQLite 用excluded表达相同语义，但需确认驱动版本 - MySQL 没有原生
ON CONFLICT，得退回到INSERT ... ON DUPLICATE KEY UPDATE方式（见下节）
MySQL 怎么写等效的 upsert 并拿到影响行数
MySQL 不支持 on_conflict_do_update()，必须用 prefix_with("ON DUPLICATE KEY UPDATE") 手动拼接。更麻烦的是：SQLAlchemy 默认不暴露 MySQL 的 ROW_COUNT()，session.execute() 返回的 Result 对象里没有“到底插入了还是更新了”的明确标识。
实操建议：
- 在
INSERT语句末尾加prefix_with("ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)")，再调用connection.exec_driver_sql("SELECT ROW_COUNT()").scalar()获取实际影响行数（1=插入，2=更新） - 更稳的方式是先
SELECT查是否存在，再分两路处理——虽然多一次查询，但逻辑清晰、可测、不依赖数据库返回值解释 - 别信
session.merge()：它内部是先查后 insert/update，且无法区分操作类型，也不保证原子性
如何让 upsert 返回完整记录（比如新生成的 id 或更新后的值）
PostgreSQL 支持 RETURNING，但必须和 on_conflict_do_update() 配合使用，且只对“最终生效的行”返回——即：插入成功时返回新行，更新成功时返回更新后的行。
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
写法要点：
-
returning()必须链在on_conflict_do_update()之后，不能放在insert()后直接调 - 返回字段要明确写，比如
returning(User.id, User.name, User.updated_at)；用User类本身会返回整行，但 ORM 映射可能不自动刷新 - 执行后用
result.fetchone()拿结果，不是scalar()——后者只适合单列
常见坑：returning() 在 SQLite 中仅限 3.35+ 且需启用 sqlite:///...?enable_returning=true 参数，否则静默失败。
ORM 层做 upsert 容易忽略的事务与刷新问题
直接用 session.execute(insert_stmt) 是最可控的，但如果你坚持用 ORM 实例（如 session.merge(user) 或 session.add(user)），要注意：
-
merge()不触发数据库级唯一冲突检测，它只基于 Python 对象 identity map 做判断，一旦并发写入就可能丢更新 -
add()+flush()后如果遇到唯一冲突，会抛IntegrityError，但此时 session 状态已脏，必须session.rollback()后重试，不能简单捕获后继续 - 即使 upsert 成功，ORM 实例的属性也不会自动同步数据库返回值（比如自增
id或服务端默认值），得手动session.refresh(instance)
真正需要 upsert 语义时，绕过 ORM 直接写 Core 语句反而更轻、更准、更容易调试。
