sqlalchemy 如何写“插入或更新”并返回 upsert 结果

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
stmt = insert(User).values(
email="alice@example.com",
name="Alice",
updated_at=func.now()
)
stmt = stmt.on_conflict_do_update(
indexelements=["email"],  # 必须匹配唯一索引/主键字段，不能写成 ["id"] 除非是主键冲突
set=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() 配合使用，且只对“最终生效的行”返回——即：插入成功时返回新行，更新成功时返回更新后的行。

写法要点：

• 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 语句反而更轻、更准、更容易调试。

# ai  # python  # 封装  # 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; ?> 】