SQLAlchemy“事件.监听”适用于所有型号

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 应用于所有模型或多个模型？

UPD：我正在尝试这样做：

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 对象时，我已 create_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)

请参阅 Mixin 和自定义基类

Answer 2

在较新版本的 sqlalchemy (1.2+) 中，以下 事件目标 可用：

• 映射类（即订阅每个模型）
• 未映射的超类（即 Base 和 mixin，使用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 类