sqlalchemy 事件 before_update 没有被执行答案

【问题标题】：sqlalchemy event before_update is not getting executedsqlalchemy 事件 before_update 没有被执行
【发布时间】：2022-08-06 20:35:42
【问题描述】：

我想维护数据库更改的审核日志（如果发生任何更新，则将先前的状态插入日志表

下面是我写的代码

from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from sqlalchemy.orm import sessionmaker

engine = create_engine(\'mysql+pymysql://root:p4ssw0rd@0.0.0.0:33061/error_details\')
Base = declarative_base()

Base.metadata.create_all(engine)
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()


class User(Base):
    __tablename__ = \"users\"
    id = Column(Integer, primary_key=True)
    name = Column(String(10))

    def __repr__(self):
        return f\"User(name={self.name}, id={self.id})\"


class User_Log(Base):
    __tablename__ = \"users_log\"
    id = Column(Integer, primary_key=True)
    name = Column(String(10))
    modified_on = Column(DateTime, default=datetime.utcnow)
    action = Column(String(10))


@event.listens_for(User, \'before_update\')
def update_user_log(mapper, connection, target):
    print(\"func() called \")


session.query(User).filter(User.id == 2).update({User.name: \"ankit--\"})
session.commit()

没有调用函数 update_user_log

注意 - 我只想在 sqlalchemy 中执行数据库审计日志

标签： python sqlalchemy

【解决方案1】：

Update and deletes with arbitrary where clauses 必须用SessionEvents.do_orm_execute_handler 拦截。这是因为

启用 ORM 的 UPDATE 和 DELETE 功能绕过 ORM 工作单元自动化，有利于能够发出单个 UPDATE 或 DELETE 语句，一次匹配多行而不复杂。*

您可以使用如下处理程序拦截此类更新：
@event.listens_for(Session, 'do_orm_execute')
def update_user_log(orm_execute_state):
    if orm_execute_state.is_update:
        print("func() called ")
* 引用来自先前链接部分末尾的第二个警告块。

【讨论】：

【解决方案2】：

在文档中不是很清楚，但merge() 将导致before_update 事件触发，请参阅Session API docs

【讨论】：