SQLAlchemy中on_conflict_do_update如何添加日期判断的更新条件？

你之前的问题出在参数用错了——index_where是用来限定哪些行参与冲突检测的，不是控制更新执行的条件。要实现「主键冲突时仅当新日期更晚才更新」的需求，得用on_conflict_do_update的where参数（SQLAlchemy 1.4+支持），结合excluded对象来对比新旧日期。

修正后的代码如下：

from sqlalchemy import excluded

def upsert_last_calc_month(model: Any, data: Dict, prim_keys: List):
    insert_stmt = insert(model).values(data)
    do_update_stmt = insert_stmt.on_conflict_do_update(
        index_elements=prim_keys,
        # 核心：用where指定更新条件，excluded指代待插入的新数据行
        where=model.date_column < excluded.date_column,
        set_=remove_key(data, "created_at"),
    )
    return do_update_stmt

关键说明： #

• excluded.date_column：代表插入语句中准备插入的date_column值，用来和数据库中已存在的model.date_column做对比
• where参数：冲突发生后，只有满足该条件时才会执行更新操作，完全匹配你的需求

如果你用的是SQLAlchemy 1.4之前的版本，可以用case表达式曲线实现（效率稍低，但效果一致）：

from sqlalchemy import case

def upsert_last_calc_month(model: Any, data: Dict, prim_keys: List):
    insert_stmt = insert(model).values(data)
    update_data = remove_key(data, "created_at")
    # 对每个字段添加条件判断，仅当新日期更晚时才更新字段
    conditional_update = {
        col: case(
            [(model.date_column < excluded.date_column, val)],
            else_=getattr(model, col)
        )
        for col, val in update_data.items()
    }
    do_update_stmt = insert_stmt.on_conflict_do_update(
        index_elements=prim_keys,
        set_=conditional_update,
    )
    return do_update_stmt

这种方式会执行更新语句，但只有符合条件时才修改字段值，不符合则保留原字段值。

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