SQLAlchemy “event.listen” for all models

Question

I have fields created_by and updated_by in each models. These fields are automatically filled with sqlalchemy.event.listen (formerly MapperExtension). For each model, I write:

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

When the model was a lot of code gets ugly. Is it possible to apply event.listen immediately to all models or several?

UPD: I'm trying to do so:

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"))

When I add a new MyClass object I have created_by = None. If I create event.listen for MyClass all is fine. What's wrong?

Answer 1

In newer versions of sqlalchemy (1.2+), the following event targets are available:

• mapped classes (that is, subscribe to every model)
• unmapped superclasses (that is, Base, and mixins, using the propagate=True flag)
• Mapper objects
• Mapper class itself

So, in order to listen to all instance events, you can listen on Mapper itself:

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)

this way you will get an app-wide event listener. If you want to only listen to your own models, use the Base class

Answer 2

Inherit all your models from the base class and subscribe to that base class:

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

See more on Mixin and Custom Base Classes