SQLAlchemy如何实现特定列变更触发的SESSION MERGE自定义逻辑？

假设数据库中已存在User记录：

[id=1, name="GRIZZLY BEAR", age=20, run_id=49019]

期望的插入/合并逻辑如下：

• 当主键id已存在时：
  • 检查除run_id外的其他列是否有变更
  • 若有变更，则合并修改内容并更新新的run_id值
• 当主键不存在时：
  • 插入新记录

示例1：不应修改数据库记录

o = User(id=1, name="GRIZZLY BEAR", age=20, run_id=49020)
session.merge(o)

示例2：应将数据库记录修改为[id=1, name="GRIZZLY BEAR", age=19, run_id=49020]

o = User(id=1, name="GRIZZLY BEAR", age=19, run_id=49020)
session.merge(o)

SQLAlchemy原生的merge会检查所有列，因此示例1中仅run_id变化时也会自动更新，不符合需求。请问是否需要自定义SQL merge查询，或是否可通过内置函数实现仅特定列变更检测的合并逻辑？

不需要完全自定义SQL查询，有两种实用方案可以实现需求：

方案一：通过SQLAlchemy事件监听自定义合并逻辑 #

利用before_merge事件，在合并前对比目标对象与数据库现有记录的指定列，仅当非run_id列有变更时才保留新的run_id，否则重置run_id为现有值，避免不必要的更新。

示例代码：

from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, "before_merge")
def before_merge(session, instance, load=True, **kwargs):
    if instance.id is None:
        return  # 主键不存在，直接走插入逻辑
    
    existing = session.query(User).get(instance.id)
    if not existing:
        return
    
    # 检查除run_id外的核心列是否有变化（可根据实际模型调整列列表）
    has_change = (existing.name != instance.name) or (existing.age != instance.age)
    
    if not has_change:
        # 无变更时，将instance的run_id同步为现有值，merge时不会触发更新
        instance.run_id = existing.run_id

配置后直接调用session.merge()即可：示例1中因为核心列无变化，run_id被重置，不会更新数据库；示例2中age有变更，会保留新的run_id并执行更新。

方案二：使用数据库原生UPSERT语句 #

直接构造数据库原生的UPSERT语句（不同数据库语法略有差异），在SQL层面控制更新条件，性能更高效。

MySQL 实现 #

from sqlalchemy import insert

stmt = insert(User).values(
    id=1, name="GRIZZLY BEAR", age=20, run_id=49020
).on_duplicate_key_update(
    name=insert(User).name,
    age=insert(User).age,
    run_id=insert(User).run_id,
    # 仅当核心列有变更时才执行更新
    _where=(User.name != insert(User).name) | (User.age != insert(User).age)
)
session.execute(stmt)
session.commit()

PostgreSQL 实现 #

from sqlalchemy import insert

stmt = insert(User).values(
    id=1, name="GRIZZLY BEAR", age=20, run_id=49020
).on_conflict_do_update(
    index_elements=['id'],  # 指定主键索引
    set_={
        'name': insert(User).name,
        'age': insert(User).age,
        'run_id': insert(User).run_id
    },
    # 核心列有变更才更新
    where=(User.name != insert(User).name) | (User.age != insert(User).age)
)
session.execute(stmt)
session.commit()

这种方式直接在SQL层控制更新逻辑，完全符合需求，且避免了ORM层面的额外查询开销。

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