如何基于SQLAlchemy事件记录提交后的增删改操作日志？

完全可以基于SQLAlchemy的事件机制实现需求，你提到的after_update/after_insert/after_delete事件的问题确实存在：这类是Mapper层级的对象事件，触发时仅代表Python对象的变更被刷入数据库执行流，事务尚未提交，一旦后续触发回滚，对应操作根本不会持久化，直接在这类事件里写日志必然会出现脏记录。

正确的实现逻辑是操作暂存+提交后落盘，核心用到Session层级的事务事件，具体实现步骤如下：

• 首先利用Mapper层的三个行级事件，仅做操作数据暂存：在事件触发时把操作类型、对应表、主键、变更内容等信息存在当前Session的自定义临时属性里，这一步不做任何日志持久化操作。
• 绑定Session的after_commit事件：该事件仅在整个事务真正提交成功、所有变更持久化到数据库后才会触发，此时取出之前暂存的所有操作记录，统一写入日志存储即可。
• 绑定Session的after_rollback事件：事务回滚触发时直接清空Session里暂存的待记录操作，避免后续新事务混入回滚过的脏操作记录。

最简实现参考代码：

from sqlalchemy.orm import Session
from sqlalchemy import event

# 初始化会话上的日志暂存区
@event.listens_for(Session, "after_transaction_create")
def init_log_buffer(session, transaction):
    if not hasattr(session, "_op_log_buffer"):
        session._op_log_buffer = []

# 示例：监听插入操作，暂存日志
@event.listens_for(YourModel, "after_insert")
def buffer_insert_log(mapper, connection, target):
    session = Session.object_session(target)
    session._op_log_buffer.append({
        "op": "INSERT",
        "table": mapper.mapped_table.name,
        "primary_key": target.id,
        "snapshot": {col.name: getattr(target, col.name) for col in mapper.columns}
    })

# 示例：监听更新操作，暂存变更字段
@event.listens_for(YourModel, "after_update")
def buffer_update_log(mapper, connection, target):
    session = Session.object_session(target)
    changes = {}
    for attr in mapper.attrs:
        history = attr.load_history(target)
        if history.has_changes():
            changes[attr.key] = (history.deleted[0] if history.deleted else None, 
                                history.added[0] if history.added else None)
    session._op_log_buffer.append({
        "op": "UPDATE",
        "table": mapper.mapped_table.name,
        "primary_key": target.id,
        "changes": changes
    })

# 示例：监听删除操作，暂存日志
@event.listens_for(YourModel, "after_delete")
def buffer_delete_log(mapper, connection, target):
    session = Session.object_session(target)
    session._op_log_buffer.append({
        "op": "DELETE",
        "table": mapper.mapped_table.name,
        "primary_key": target.id
    })

# 事务提交成功后持久化日志
@event.listens_for(Session, "after_commit")
def persist_logs(session):
    if not hasattr(session, "_op_log_buffer") or not session._op_log_buffer:
        return
    # 替换为实际的日志写入逻辑：写日志表、推日志服务等
    for log_item in session._op_log_buffer:
        print(f"persist operation log: {log_item}")
    # 写完清空缓冲区
    session._op_log_buffer.clear()

# 回滚时清空缓冲区的脏日志
@event.listens_for(Session, "after_rollback")
def clear_buffer_on_rollback(session):
    if hasattr(session, "_op_log_buffer"):
        session._op_log_buffer.clear()

注意：如果你的操作日志需要存储在同一个数据库的日志表中，持久化日志时请新建独立的数据库会话执行写入，不要复用触发after_commit的当前会话，避免出现会话状态异常。

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