SQLAlchemy 为所有模型使用 "event.listen

Question

我在每个模型里都有两个字段，分别是 created_by 和 updated_by。这些字段会自动填充，使用的是 sqlalchemy.event.listen（以前叫 MapperExtension）。对于每个模型，我都会写一些代码：

event.listen(Equipment, 'before_insert', get_created_by_id)
event.listen(Equipment, 'before_update', get_updated_by_id)

但是当模型的代码很多的时候，看起来就很乱。有没有办法能一次性对所有模型或者几个模型应用 event.listen 呢？

更新：我现在尝试这样做：

import pylons
from sqlalchemy import event, sql
from sqlalchemy import Table, ForeignKey, Column
from sqlalchemy.databases import postgresql
from sqlalchemy.schema import UniqueConstraint, CheckConstraint
from sqlalchemy.types import String, Unicode, UnicodeText, Integer, DateTime,\
                             Boolean, Float
from sqlalchemy.orm import relation, backref, synonym, relationship
from sqlalchemy import func
from sqlalchemy import desc
from sqlalchemy.orm.exc import NoResultFound

from myapp.model.meta import Session as s
from myapp.model.meta import metadata, DeclarativeBase

from pylons import request

def created_by(mapper, connection, target):
    identity = request.environ.get('repoze.who.identity')
    if identity:
        id = identity['user'].user_id
        target.created_by = id

def updated_by(mapper, connection, target):
    identity = request.environ.get('repoze.who.identity')
    if identity:
        id = identity['user'].user_id
        target.updated_by = id

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.declarative import has_inherited_table

class TestMixin(DeclarativeBase):
    __tablename__ = 'TestMixin'

    id =  Column(Integer, autoincrement=True, primary_key=True)

event.listen(TestMixin, 'before_insert', created_by)
event.listen(TestMixin, 'before_update', updated_by)

class MyClass(TestMixin):
    __tablename__ = 'MyClass'
    __mapper_args__ = {'concrete':True}

    id =  Column(Integer, autoincrement=True, primary_key=True)

    created_by = Column(Integer, ForeignKey('user.user_id',
                        onupdate="cascade", ondelete="restrict"))

    updated_by = Column(Integer, ForeignKey('user.user_id',
                        onupdate="cascade", ondelete="restrict"))

当我添加一个新的 MyClass 对象时，created_by 的值是 None。如果我为 MyClass 创建了 event.listen，那一切都正常。那到底出了什么问题呢？

Answer 1

把你所有的模型都从一个基础类继承，并且要订阅这个基础类：

event.listen(MyBaseMixin, 'before_insert', get_created_by_id, propagate=True)
event.listen(MyBaseMixin, 'before_update', get_updated_by_id, propagate=True)

想了解更多，可以查看 混合类和自定义基础类

Answer 2

在较新版本的sqlalchemy（1.2及以上）中，以下这些事件目标可以使用：

• 映射类（也就是说，订阅每一个模型）
• 未映射的超类（也就是Base和混合类，使用propagate=True标志）
• Mapper对象
• Mapper类本身

所以，为了监听所有实例事件，你可以直接在Mapper上进行监听：

from typing import Set, Optional

import sqlalchemy as sa
import sqlalchemy.orm.query
import sqlalchemy.event

@sa.event.listens_for(sa.orm.Mapper, 'refresh', named=True)
def on_instance_refresh(target: type, 
                        context: sa.orm.query.QueryContext, 
                        attrs: Optional[Set[str]]):
    ssn: sqlalchemy.orm.Session = context.session
    print(target, attrs)

这样你就能得到一个全应用的事件监听器。如果你只想监听自己的模型，可以使用Base类。