python - 在所有模型之间共享事件处理程序

我试图在所有模型中共享一个简单的功能（时间戳记），并采用SQLA文档中所述的“增强基础”方法。到目前为止，我做到了：

import sqlalchemy as sa
from datetime import datetime as dt

class EntityBase(object):

    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    id = sa.Column(sa.Integer, primary_key=True)
    last_update = sa.Column(sa.DateTime, default=dt.utcnow())

def update_entity(mapper, connection, target):
    target.last_update = dt.utcnow()


Entity = declarative_base(cls=EntityBase)

sa.event.listen(Entity, 'before_insert', update_entity)
sa.event.listen(Entity, 'before_update', update_entity)

我所有的模型均来自Entity类。但是在运行时我得到了sqlalchemy.orm.exc.UnmappedClassError: Class 'sqlalchemy.ext.declarative.Base' is not mapped。我究竟做错了什么？

更新
我通过执行简单的预处理来避免了这个问题，如下所示：

def setupEntities():
    ...
    for cls in Entity.__subclasses__():
        listen(cls, 'before_insert', update_entity)
        listen(cls, 'before_update', update_entity)
    ...

...但是我想听听正确的方法。

最佳答案

有趣的是，上面的操作方式可能应该起作用。会很方便。因此，我为该http://www.sqlalchemy.org/trac/ticket/2585添加了票证。

现在，您可以通过设置Base的新映射事件监听器来实现：

Entity = declarative_base(cls=EntityBase):

from sqlalchemy.orm import mapper
@event.listens_for(mapper, 'mapper_configured')
def set_events(mapper, class_):
   if issubclass(class_, Entity):
        listen(class_, 'before_update', update_entity)
        listen(class_, 'before_insert', update_entity)

实际上，如果我为＃2585实现该功能，则可能也必须与此非常类似。

关于python - 在所有模型之间共享事件处理程序，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/12640469/