SQLAlchemy中on_conflict_do_update如何追加更新冲突行的数组字段

ORM 模型定义 #

class Example(Base):
    __tablename__ = "example"

    create_time = Column(DateTime, server_default=func.now())
    time_stamps = Column(MutableList.as_mutable(ARRAY(DateTime)), server_default="{}")
    update_time = Column(DateTime, server_default=func.now())

业务规则 #

插入Example表新记录时，触发唯一约束冲突需要满足的处理逻辑：

• 将本次新记录的create_time追加到冲突行的time_stamps数组
• 对追加完成的数组按时间倒序排序
• 取排序后数组的最新时间，作为冲突行的新update_time值
• 整个流程需要实现数据库层原子操作，规避分步查询更新的并发一致性问题

原有实现 #

之前已经落地的非原子更新逻辑：先查询到对应的冲突记录，在内存中修改字段值后再提交事务，无法保证操作的原子性：

def update_record(db: Session, create_time: datetime, db_record: Example):
    db_record.time_stamps.append(create_time)
    sorted_times = sorted(db_record.time_stamps, reverse=True)
    db_record.update_time = sorted_times[0]
    db_record.time_stamps = sorted_times
    db.commit()

当前开发卡点 #

现在需要改用INSERT ON CONFLICT UPDATE语法实现原子操作，已经完成基础插入逻辑编写，但在配置冲突更新的字段赋值规则时无法推进：

db_dict = {
    "create_time": record.create_time,
    "time_stamps": [record.create_time],
    "update_time": record.create_time
}

stm = insert(Example).values(db_dict)
do_update_stm = stm.on_conflict_do_update(
    constraint='my_unique_constraint',
    set_=dict(??)  # 字段赋值规则未实现
)

核心问题 #

在SQLAlchemy的on_conflict_do_update的set_参数中，如何访问原有冲突行的字段值，实现time_stamps数组的追加、排序操作？ #

解决方案 #

核心语法说明 #

• 读取原冲突行的字段值：直接引用ORM模型对应的Column属性即可，比如Example.time_stamps就代表原表冲突行的time_stamps字段值
• 读取本次尝试插入的新值：SQLAlchemy的PostgreSQL方言提供了excluded对象，对应原生SQL里ON CONFLICT语法中的EXCLUDED行，也就是本次本该插入但因为冲突被跳过的新记录

完整可运行实现 #

结合PostgreSQL原生数组操作函数，完成数组追加、排序、字段同步更新的逻辑：

from sqlalchemy.dialects.postgresql import excluded
from sqlalchemy import func, select

# 定义冲突触发时的字段赋值规则
update_values = {
    # 1. 数组追加：把原行time_stamps和本次插入的create_time拼接
    # 2. 对拼接后的数组按时间倒序排序
    "time_stamps": func.array(
        select(
            func.unnest(
                func.array_append(Example.time_stamps, excluded.create_time)
            )
        ).order_by(
            func.unnest(func.array_append(Example.time_stamps, excluded.create_time)).desc()
        )
    ),
    # 同步更新最新时间：因为本次追加的create_time是当前操作的时间戳，天然是数组中最新的值，直接取值即可
    "update_time": excluded.create_time
}

do_update_stm = stm.on_conflict_do_update(
    constraint='my_unique_constraint',
    set_=update_values
)

提示：如果担心业务上出现时间倒挂的情况（比如传入的create_time早于数组已有值），可以把update_time改成取排序后数组的第一个元素：func.coalesce(Example.time_stamps[0], excluded.create_time)，保证时间值的正确性。

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