基于PostgreSQL的SqlAlchemy批量Upsert：为冲突行设置唯一值

嘿，这个需求我之前做项目时刚好遇到过！完全不用写for循环，靠SQLAlchemy的on_conflict_do_update配合PostgreSQL的excluded对象就能搞定一条SQL实现批量Upsert，而且每行冲突时都会用自己的新值更新。

核心思路 #

PostgreSQL的ON CONFLICT子句允许我们在插入冲突（比如主键/唯一键重复）时执行更新操作，而SQLAlchemy的excluded对象可以引用当前待插入行的字段值——这正是实现“每行更新专属值”的关键，它会自动对应到冲突的那行数据，不用手动指定。

具体代码示例 #

假设我们有一个简单的用户模型：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100), unique=True)

接下来是批量Upsert的代码：

from sqlalchemy import insert
from sqlalchemy.exc import SQLAlchemyError
from your_module import User, db  # 这里的db是你的SQLAlchemy会话实例

# 准备批量数据，每个字典对应一行要Upsert的记录
batch_records = [
    {"id": 1, "name": "Cody Updated", "email": "cody_updated@example.com"},
    {"id": 2, "name": "New User Alice", "email": "alice@example.com"},
    {"id": 3, "name": "Bob Updated", "email": "bob_new@example.com"}
]

try:
    # 构造批量插入语句
    insert_stmt = insert(User).values(batch_records)
    
    # 定义冲突处理逻辑：当id（主键）冲突时，更新name和email为当前待插入行的值
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["id"],  # 指定触发冲突的唯一键/主键
        set_={
            # 用excluded引用当前待插入行的字段值
            "name": insert_stmt.excluded.name,
            "email": insert_stmt.excluded.email
        }
    )
    
    # 执行并提交
    db.session.execute(upsert_stmt)
    db.session.commit()
    print("批量Upsert完成！")
except SQLAlchemyError as e:
    db.session.rollback()
    print(f"操作出错：{str(e)}")

关键细节解释 #

1. index_elements：这里指定的是用来判断冲突的字段（可以是主键、唯一约束列，甚至是多个列组成的复合唯一约束），比如如果你的冲突判断是基于email唯一键，就把这里改成["email"]。
2. insert_stmt.excluded：这个对象是PostgreSQL提供的特殊引用，它代表当前因为冲突而无法插入的那行数据。所以每个冲突行都会自动用自己的新值去更新，而不是所有行都用同一个固定值。
3. 复杂更新逻辑：如果需要更复杂的更新（比如拼接字符串、数值计算），可以结合SQLAlchemy的函数，比如：

   from sqlalchemy import func
   set_={
       "name": func.concat(insert_stmt.excluded.name, " (updated)"),
       "email": insert_stmt.excluded.email
   }
   

注意事项 #

• 确保你的PostgreSQL版本在9.5以上（支持ON CONFLICT），SQLAlchemy版本在1.3以上（支持on_conflict_do_update语法）。
• 表必须有对应的唯一约束或主键，否则ON CONFLICT不会触发。

内容的提问来源于stack exchange，提问作者Cody Moncur