SQLAlchemy:在插入或更新之前设置审计列

Question

我的每个映射类都包含created_by和updated_by审计属性，我希望在各自对象的INSERT和UPDATE时自动设置这些属性。

class User(Base):
__tablename__ = 'user'
id = Column(BigInteger, primary_key=True)
name = Column(Text, nullable=False)
...
class Address(Base):
__tablename__ = 'address'
id = Column(BigInteger, primary_key=True)
street = Column(Text, nullable=False)
...
created_by = Column(BigInteger)  # references user.id
updated_by = Column(BigInteger)  # references user.id
...

是否有办法在SQLAlchemy中集中处理这个问题?我查看了事件，但似乎需要为每个单独的映射类单独设置(注意装饰器中的SomeClass)。

@event.listens_for(SomeClass, 'before_insert')
def on_insert(mapper, connection, target):
target.created_by = context["current_user"] # I want to be able to do this not just for 'SomeClass' but for all mapped classes

@event.listens_for(SomeClass, 'before_update')
def on_update(mapper, connection, target):
target.updated_by = context["current_user"] # I want to be able to do this not just for 'SomeClass' but for all mapped classes

Answer 1

这里的一个解决方案是使用sqlalchemy提供的Column类中的默认参数。实际上，你可以传递一个可调用对象给default(在第一次创建时执行)和onupdate(在每次更新时执行)。

def get_current_user():
return context["user"].id

class Address(Base):
__tablename__ = 'address'
...
created_by = Column(default = get_current_user)
updated_by = Column(default = get_current_user, onupdate=get_current_user)

Answer 2

设法弄清楚了，虽然有点担心在declarative_base上使用dunder方法__subclasses__()。如果有更好的选择，请提出建议。

def on_insert(mapper, connection, target):
target.created_by = context["user"].id
target.updated_at = datetime.utcnow()

def on_update(mapper, connection, target):
target.updated_by = context["user"].id
target.updated_at = datetime.utcnow()

Base.metadata.create_all()
mapped_classes = Base.__subclasses__()
for mapped_class in mapped_classes:
event.listen(mapped_class, 'before_insert', on_insert)
event.listen(mapped_class, 'before_update', on_update)

这里所指的上下文实际上是starletter -context