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

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对象时,我创建了_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和Custom Base Classes的更多信息,请参阅

Answer 2

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

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