Flask-SQLAlchemy实现user表增改时写入user_log审计表的最佳方案

问题背景 #

我拥有两张数据表：

1. user - 用户详情表
2. user_log - 审计表，用于记录某一时刻的用户详情。即当user表中特定用户发生插入或更新操作时，user_log需保存该用户的旧记录。

表定义如下：

class User(db.Model):
    user_id = db.Column(db.Integer, primary_key = True)
    name = db.Column(db.String, nullable = False)
    address=  db.Column(db.String, nullable = False)

class UserLog(db.Model):
    id = db.Column(db.Integer, primary_key = True)
    user_id = db.Column(db.Integer,  nullable = False)
    name = db.Column(db.String, nullable = False)
    address=  db.Column(db.String, nullable = False)

我曾尝试使用MapperEvents，但调用update()方法时该方案无法生效，请问实现该功能的最佳方式是什么？

解决方案 #

方案一：实例级事件监听（仅支持对象属性修改） #

你之前用MapperEvents失效的核心原因是：直接调用User.query.filter(...).update(...)这类批量更新操作不会触发实例级事件，只有通过修改User实例属性再提交的方式才会触发。

代码实现：

from sqlalchemy import event

def log_old_user_on_update(mapper, connection, target):
    # 从数据库获取更新前的旧数据（加锁避免并发问题）
    old_user = db.session.query(User).with_for_update().get(target.user_id)
    if old_user:
        log_entry = UserLog(
            user_id=old_user.user_id,
            name=old_user.name,
            address=old_user.address
        )
        db.session.add(log_entry)

def log_user_on_insert(mapper, connection, target):
    # 插入时记录初始状态（可根据需求选择是否保留）
    log_entry = UserLog(
        user_id=target.user_id,
        name=target.name,
        address=target.address
    )
    db.session.add(log_entry)

# 绑定事件到User模型
event.listen(User, 'before_update', log_old_user_on_update)
event.listen(User, 'before_insert', log_user_on_insert)

方案二：数据库触发器（支持所有更新操作） #

如果需要覆盖批量更新、直接SQL操作等场景，数据库触发器是最可靠的方案。SQLAlchemy可以通过DDL语句在表创建后自动生成触发器。

以MySQL为例：

from sqlalchemy import DDL

# 更新触发器：更新前插入旧记录到user_log
update_trigger = DDL("""
CREATE TRIGGER user_before_update_trigger BEFORE UPDATE ON user
FOR EACH ROW
BEGIN
    INSERT INTO user_log (user_id, name, address)
    VALUES (OLD.user_id, OLD.name, OLD.address);
END;
""")

# 插入触发器：插入时记录初始状态（可根据需求选择）
insert_trigger = DDL("""
CREATE TRIGGER user_before_insert_trigger BEFORE INSERT ON user
FOR EACH ROW
BEGIN
    INSERT INTO user_log (user_id, name, address)
    VALUES (NEW.user_id, NEW.name, NEW.address);
END;
""")

# 绑定触发器到User表的创建事件
event.listen(User.__table__, 'after_create', update_trigger)
event.listen(User.__table__, 'after_create', insert_trigger)

注意：不同数据库的触发器语法有差异，比如PostgreSQL、SQL Server需要调整对应的语法。

方案三：自定义更新方法（可控性强） #

封装统一的更新方法，强制在更新前记录旧数据，适合需要严格控制更新流程的场景：

@classmethod
def update_and_log(cls, user_id, **kwargs):
    old_user = cls.query.get(user_id)
    if not old_user:
        return None
    
    # 记录旧记录到审计表
    log_entry = UserLog(
        user_id=old_user.user_id,
        name=old_user.name,
        address=old_user.address
    )
    db.session.add(log_entry)
    
    # 执行更新
    for key, value in kwargs.items():
        if hasattr(old_user, key):
            setattr(old_user, key, value)
    
    db.session.commit()
    return old_user

使用时统一调用User.update_and_log(1, name="新名称", address="新地址")，确保所有更新操作都经过日志记录环节。

方案对比 #

• 实例级事件：实现简单，但仅支持实例属性修改的场景，无法覆盖批量更新。
• 数据库触发器：覆盖所有操作场景，无需修改业务代码，但依赖数据库语法，跨库兼容性差。
• 自定义更新方法：可控性强，便于扩展，但需要团队统一遵循调用规范，无法拦截直接的SQL操作。

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