SQLAlchemy中通过do_orm_execute事件监听器获取更新值的最佳方案

问题背景 #

在使用SQLAlchemy事件监听器时，希望通过do_orm_execute会话事件获取执行execute()语句时的实际更新值，类似用before_flush检测ORM事件的实现方式：

def _before_flush(session, flush_context, instances):
    for entry in session.dirty:
        # 打印更新后的数据表行类实例
        print(entry)
event.listen(session, "before_flush", _before_flush)

需要找到近似方法处理do_orm_execute事件，现有代码框架如下：

def _receive_orm_execute(orm_execute_state: ORMExecuteState):
    if orm_execute_state.is_update:
        # 此处如何获取更新值？
        # 发现属性orm_execute_state.execution_options['_sa_orm_update_options']._resolved_keys_as_propnames似乎包含更新字段值的元组，但不确定是否该使用它
event.listen(session, "do_orm_execute", _receive_orm_execute)

核心需求：获取execute()执行更新操作时实际更新值的最佳/推荐方式（基于SQLAlchemy v1.4.x）

解决方案 #

针对SQLAlchemy 1.4.x版本，推荐以下几种可靠方式获取更新值：

1. 从ORM更新语句中提取绑定参数 #

如果更新是通过ORM的update()构造器实现（比如session.execute(update(User).where(User.id==1).values(name="new_name"))），可以直接从编译后的语句中提取参数，这是官方推荐的安全方式：

def _receive_orm_execute(orm_execute_state: ORMExecuteState):
    if orm_execute_state.is_update:
        # 获取更新语句的绑定参数
        compiled_stmt = orm_execute_state.statement.compile()
        update_params = compiled_stmt.params
        print("ORM更新参数：", update_params)

2. 处理原生SQL更新的参数 #

如果是执行原生SQL更新（比如session.execute(text("UPDATE user SET name=:name WHERE id=1"), {"name": "new_name"})），可直接通过orm_execute_state.parameters获取传入的参数：

def _receive_orm_execute(orm_execute_state: ORMExecuteState):
    if orm_execute_state.is_update:
        # 获取原生SQL的更新参数
        native_update_params = orm_execute_state.parameters
        print("原生SQL更新参数：", native_update_params)

3. 内部私有属性（谨慎使用） #

你提到的_sa_orm_update_options属于SQLAlchemy内部私有属性，虽然能提取更新字段信息，但官方不保证后续版本兼容性，仅在特殊场景下临时使用：

def _receive_orm_execute(orm_execute_state: ORMExecuteState):
    if orm_execute_state.is_update:
        update_options = orm_execute_state.execution_options.get('_sa_orm_update_options')
        if update_options:
            # 获取更新字段与对应值的映射
            update_values = update_options.values
            print("更新字段及值：", update_values)

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