SQLAlchemy如何全局配置插入更新时自动设置审计字段

SQLAlchemy完全支持全局统一处理这类审计字段自动赋值的需求，不需要为每个映射类单独注册事件监听。

方案1：全局Mapper事件监听（无需修改现有模型定义） #

你不需要把事件绑定到具体的模型类，直接将监听目标设置为Mapper基类，加上propagate=True参数，就能让监听逻辑对所有映射类生效。只需要加一层属性判断，避免给没有对应审计字段的模型赋值报错即可。

示例代码：

from sqlalchemy import event
from sqlalchemy.orm import Mapper

# 全局监听所有模型的INSERT操作
@event.listens_for(Mapper, "before_insert", propagate=True)
def auto_set_created_by(mapper, connection, target):
    # 仅当模型存在created_by字段时才赋值
    if hasattr(target, "created_by"):
        target.created_by = context["current_user"]

# 全局监听所有模型的UPDATE操作
@event.listens_for(Mapper, "before_update", propagate=True)
def auto_set_updated_by(mapper, connection, target):
    # 仅当模型存在updated_by字段时才赋值
    if hasattr(target, "updated_by"):
        target.updated_by = context["current_user"]

这个方案不需要改动你现有的模型代码，写完两个监听函数之后，所有带对应字段的模型在执行插入、更新时都会自动填充用户ID。

方案2：Mixin封装（适合新项目，代码更易维护） #

如果你的项目里审计字段是通用需求，可以用Mixin类把字段定义和逻辑判断做统一封装，避免重复写字段定义，逻辑也更严谨：

from sqlalchemy import Column, BigInteger, Text, event
from sqlalchemy.orm import Mapper, declarative_base

Base = declarative_base()

# 定义审计字段通用Mixin
class AuditFieldsMixin:
    created_by = Column(BigInteger, comment="创建人ID")
    updated_by = Column(BigInteger, comment="更新人ID")

# 不需要审计字段的模型保持原有写法即可
class User(Base):
    __tablename__ = 'user'
    id = Column(BigInteger, primary_key=True)
    name = Column(Text, nullable=False)

# 需要审计字段的模型多继承一个Mixin即可，不用重复写字段定义
class Address(Base, AuditFieldsMixin):
    __tablename__ = 'address'
    id = Column(BigInteger, primary_key=True)
    street = Column(Text, nullable=False)

# 监听逻辑可以用类型判断替代属性判断，更严谨
@event.listens_for(Mapper, "before_insert", propagate=True)
def auto_set_created_by(mapper, connection, target):
    if isinstance(target, AuditFieldsMixin):
        target.created_by = context["current_user"]

@event.listens_for(Mapper, "before_update", propagate=True)
def auto_set_updated_by(mapper, connection, target):
    if isinstance(target, AuditFieldsMixin):
        target.updated_by = context["current_user"]

注意事项 #

• 存储current_user的context对象请优先使用Python标准库的contextvars模块实现，在Web框架的中间件/请求钩子中完成用户ID的赋值和清理，避免多线程、协程场景下出现用户数据串扰的问题。
• 如果你使用的是SQLAlchemy 2.0+版本，上述写法完全兼容，不需要做额外调整。
• 文档中单个类绑定事件的示例只是演示基础用法，事件系统本身支持基类级别的全局监听，propagate=True就是用来控制事件是否传播到所有子类/映射实例的参数。

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