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



我正在尝试在我的所有模型之间共享一个简单的功能(时间戳),并采用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 的功能,我可能也必须这样做。

最新更新