如何从SQLAlchemy的do_orm_execute事件获取目标实体实现历史表？

方案可行性结论 #

完全可行，do_orm_execute事件可以覆盖ORM操作和直接操作Table的Core操作，只要正确解析执行状态中的信息即可。

核心解决方法 #

1. 从ORMExecuteState提取目标实体/表 #

不管是ORM的update/delete还是直接操作Table的Core操作，都能从ORMExecuteState.statement中解析出目标对象：

• ORM操作：通过statement.target直接拿到对应的ORM模型类，后续用issubclass判断是否需要记录历史。
• Core操作：statement本身是Update或Delete对象，statement.table就是对应的Table实例，若该Table已绑定declarative_base的模型，可通过table.class_关联到ORM模型；未绑定的话，也可通过table.name匹配需要记录历史的表。

2. 代码示例 #

from sqlalchemy import Update, Delete
from sqlalchemy import event
from sqlalchemy.orm import ORMExecuteState

def handle_history_logs(state: ORMExecuteState):
    # 只处理UPDATE和DELETE操作
    if not isinstance(state.statement, (Update, Delete)):
        return

    target_entity = None
    # 区分ORM操作与Core操作
    if hasattr(state.statement, 'target'):
        # ORM操作，获取模型类
        target_entity = state.statement.target
    elif hasattr(state.statement, 'table'):
        # Core操作，先获取Table，再关联到ORM模型（如果已绑定）
        target_table = state.statement.table
        target_entity = target_table.class_ if hasattr(target_table, 'class_') else None

    # 判断是否为需要记录历史的模型（假设日志Mixin为HistoryMixin）
    if target_entity and issubclass(target_entity, HistoryMixin):
        # 这里实现历史记录逻辑：
        # ORM操作可通过state.session获取变更实例；Core操作可解析statement.values()或查询变更前数据
        # 示例：插入历史表
        # history_entry = HistoryTable(
        #     table_name=target_entity.__tablename__,
        #     record_id=...,  # 根据操作条件提取ID
        #     change_data=...,  # 提取变更内容
        #     operation_type=state.statement.__class__.__name__
        # )
        # state.session.add(history_entry)

# 绑定事件到异步Session
event.listen(async_session, 'do_orm_execute', handle_history_logs)

3. 兼容逐步迁移场景 #

这个方案能同时覆盖两种操作场景：

• 已迁移的ORM模型：直接通过target_entity判断是否继承日志Mixin
• 未迁移的Core Table：通过table.name匹配目标表，手动编写历史表插入逻辑

4. 异步场景注意事项 #

如果事件处理函数需要执行数据库操作，必须使用异步调用，比如await state.session.execute(...)，避免阻塞事件循环。

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