SQLAlchemy中从表到模型的反向映射

Question

为了在我的基于 SQLAlchemy 的应用程序中提供活动日志，我有一个这样的模型：

class ActivityLog(Base):
    __tablename__ = 'activitylog'
    id = Column(Integer, primary_key=True)
    activity_by_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    activity_by = relation(User, primaryjoin=activity_by_id == User.id)
    activity_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    activity_type = Column(SmallInteger, nullable=False)

    target_table = Column(Unicode(20), nullable=False)
    target_id = Column(Integer, nullable=False)
    target_title = Column(Unicode(255), nullable=False)

日志包含多个表的条目，所以我不能使用 ForeignKey 关系。日志条目是这样制作的：

doc = Document(name=u'mydoc', title=u'My Test Document',
               created_by=user, edited_by=user)
session.add(doc)
session.flush() # See note below
log = ActivityLog(activity_by=user, activity_type=ACTIVITY_ADD,
                  target_table=Document.__table__.name, target_id=doc.id,
                  target_title=doc.title)
session.add(log)

这给我留下了三个问题：

1. 我必须在我的doc 对象获得id 之前刷新会话。如果我使用了 ForeignKey 列和 relation 映射器，我可以简单地调用 ActivityLog(target=doc) 并让 SQLAlchemy 完成工作。有什么办法可以解决需要手动冲洗的问题吗？

2. target_table 参数过于冗长。我想我可以使用 ActivityLog 中的 target 属性设置器来解决这个问题，该设置器会自动从给定实例中检索表名和 id。

3. 最重要的是，我不确定如何从数据库中检索模型实例。给定一个 ActivityLog 实例log，调用self.session.query(log.target_table).get(log.target_id) 不起作用，因为query() 需要一个模型作为参数。

一种解决方法似乎是使用多态性并从 ActivityLog 识别的基本模型派生我的所有模型。像这样的：

class Entity(Base):
    __tablename__ = 'entities'
    id = Column(Integer, primary_key=True)
    title = Column(Unicode(255), nullable=False)
    edited_at = Column(DateTime, onupdate=datetime.utcnow, nullable=False)
    entity_type = Column(Unicode(20), nullable=False)
    __mapper_args__ = {'polymorphic_on': entity_type}

class Document(Entity):
    __tablename__ = 'documents'
    __mapper_args__ = {'polymorphic_identity': 'document'}
    body = Column(UnicodeText, nullable=False)

class ActivityLog(Base):
    __tablename__ = 'activitylog'
    id = Column(Integer, primary_key=True)
    ...
    target_id = Column(Integer, ForeignKey('entities.id'), nullable=False)
    target = relation(Entity)

如果我这样做，ActivityLog(...).target 将在引用 Document 时给我一个 Document 实例，但我不确定是否值得为所有内容创建两个表的开销。我应该继续这样做吗？

Answer 1

解决这个问题的一种方法是多态关联。它应该可以解决您的所有 3 个问题，并且还可以使数据库外键约束起作用。请参阅 SQLAlchemy 源代码中的 polymorphic association example。 Mike Bayer 有一个旧的blogpost 对此进行了更详细的讨论。

Answer 2

一定要浏览博文和 Ants 链接到的示例。我没有发现解释混乱，而是假设在该主题上有更多经验。

我可以建议的几件事是：

• ForeignKeys：总的来说，我同意它们是一件好事，但我不确定在你的情况下它在概念上是否重要：你似乎将这个ActivityLog 用作正交横切关注点（AOP）；但是带有外键的版本会有效地让您的业务对象意识到ActivityLog。使用架构设置将 FK 用于审计目的的另一个问题是，如果您允许对象 deletion，FK 要求将删除该对象的所有 ActivityLog 条目。
• Automatic logging：每当您创建/修改（/删除）对象时，您都在手动执行所有这些日志记录。使用 SA，您可以使用 before_commit 实现 SessionExtension，它会自动为您完成这项工作。

这样你就完全可以避免像下面这样写部分：

log = ActivityLog(activity_by=user, activity_type=ACTIVITY_ADD,
                  target_table=Document.__table__.name, target_id=doc.id,
                  target_title=doc.title)
session.add(log)

---

EDIT-1：添加了完整的示例代码

• 代码基于http://techspot.zzzeek.org/?p=13的第一个非FK版本。
• 选择不使用 FK 是基于这样一个事实，即出于审计目的，当 主对象被删除，它不应该级联删除所有的审计日志条目。 这也使可审计对象不知道他们正在被审计的事实。
• 实施使用 SA 一对多关系。有可能是一些 对象被多次修改，这将导致许多审计日志条目。 默认情况下，SA 将在添加新条目时加载关系对象 列表。假设在“正常”使用期间，我们只想添加新的审计 日志条目，我们使用lazy='noload' 标志，以便主对象的关系 永远不会被加载。虽然从另一侧导航时会加载它， 也可以使用自定义查询从主对象加载，如图所示 在示例中也使用了activitylog_readonly readonly 属性。

代码（可通过一些测试运行）：

from datetime import datetime

from sqlalchemy import create_engine, Column, Integer, SmallInteger, String, DateTime, ForeignKey, Table, UnicodeText, Unicode, and_
from sqlalchemy.orm import relationship, dynamic_loader, scoped_session, sessionmaker, class_mapper, backref
from sqlalchemy.orm.session import Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm.interfaces import SessionExtension

import logging
logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger()

ACTIVITY_ADD = 1
ACTIVITY_MOD = 2
ACTIVITY_DEL = 3

class ActivityLogSessionExtension(SessionExtension):
    _logger = logging.getLogger('ActivityLogSessionExtension')

    def before_commit(self, session):
        self._logger.debug("before_commit: %s", session)
        for d in session.new:
            self._logger.info("before_commit >> add: %s", d)
            if hasattr(d, 'create_activitylog'):
                log = d.create_activitylog(ACTIVITY_ADD)
        for d in session.dirty:
            self._logger.info("before_commit >> mod: %s", d)
            if hasattr(d, 'create_activitylog'):
                log = d.create_activitylog(ACTIVITY_MOD)
        for d in session.deleted:
            self._logger.info("before_commit >> del: %s", d)
            if hasattr(d, 'create_activitylog'):
                log = d.create_activitylog(ACTIVITY_DEL)


# Configure test data SA
engine = create_engine('sqlite:///:memory:', echo=False)
session = scoped_session(sessionmaker(bind=engine, autoflush=False, extension=ActivityLogSessionExtension()))
Base = declarative_base()
Base.query = session.query_property()

class _BaseMixin(object):
    """ Just a helper mixin class to set properties on object creation.  
    Also provides a convenient default __repr__() function, but be aware that 
    also relationships are printed, which might result in loading relations.
    """
    def __init__(self, **kwargs):
        for k,v in kwargs.items():
            setattr(self, k, v)

    def __repr__(self):
        return "<%s(%s)>" % (self.__class__.__name__, 
            ', '.join('%s=%r' % (k, self.__dict__[k]) 
                for k in sorted(self.__dict__) if '_sa_' != k[:4] and '_backref_' != k[:9])
            )

class User(Base, _BaseMixin):
    __tablename__ = u'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Document(Base, _BaseMixin):
    __tablename__ = u'documents'
    id = Column(Integer, primary_key=True)
    title = Column(Unicode(255), nullable=False)
    body = Column(UnicodeText, nullable=False)

class Folder(Base, _BaseMixin):
    __tablename__ = u'folders'
    id = Column(Integer, primary_key=True)
    title = Column(Unicode(255), nullable=False)
    comment = Column(UnicodeText, nullable=False)

class ActivityLog(Base, _BaseMixin):
    __tablename__ = u'activitylog'
    id = Column(Integer, primary_key=True)

    activity_by_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    activity_by = relationship(User) # @note: no need to specify the primaryjoin
    activity_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    activity_type = Column(SmallInteger, nullable=False)

    target_table = Column(Unicode(20), nullable=False)
    target_id = Column(Integer, nullable=False)
    target_title = Column(Unicode(255), nullable=False)
    # backref relation for auditable
    target = property(lambda self: getattr(self, '_backref_%s' % self.target_table))

def _get_user():
    """ This method returns the User object for the current user.
    @todo: proper implementation required
    @hack: currently returns the 'user2'
    """
    return session.query(User).filter_by(name='user2').one()

# auditable support function
# based on first non-FK version from http://techspot.zzzeek.org/?p=13
def auditable(cls, name):
    def create_activitylog(self, activity_type):
        log = ActivityLog(activity_by=_get_user(),
                          activity_type=activity_type,
                          target_table=table.name, 
                          target_title=self.title,
                          )
        getattr(self, name).append(log)
        return log

    mapper = class_mapper(cls)
    table = mapper.local_table
    cls.create_activitylog = create_activitylog

    def _get_activitylog(self):
        return Session.object_session(self).query(ActivityLog).with_parent(self).all()
    setattr(cls, '%s_readonly' %(name,), property(_get_activitylog))

    # no constraints, therefore define constraints in an ad-hoc fashion.
    primaryjoin = and_(
            list(table.primary_key)[0] == ActivityLog.__table__.c.target_id,
            ActivityLog.__table__.c.target_table == table.name
    )
    foreign_keys = [ActivityLog.__table__.c.target_id]
    mapper.add_property(name, 
            # @note: because we use the relationship, by default all previous
            # ActivityLog items will be loaded for an object when new one is
            # added. To avoid this, use either dynamic_loader (http://www.sqlalchemy.org/docs/reference/orm/mapping.html#sqlalchemy.orm.dynamic_loader)
            # or lazy='noload'. This is the trade-off decision to be made.
            # Additional benefit of using lazy='noload' is that one can also
            # record DEL operations in the same way as ADD, MOD
            relationship(
                ActivityLog,
                lazy='noload',  # important for relationship
                primaryjoin=primaryjoin, 
                foreign_keys=foreign_keys,
                backref=backref('_backref_%s' % table.name, 
                    primaryjoin=list(table.primary_key)[0] == ActivityLog.__table__.c.target_id, 
                    foreign_keys=foreign_keys)
        )
    )

# this will define which classes support the ActivityLog interface
auditable(Document, 'activitylogs')
auditable(Folder, 'activitylogs')

# create db schema
Base.metadata.create_all(engine)


## >>>>> TESTS >>>>>>

# create some basic data first
u1 = User(name='user1')
u2 = User(name='user2')
session.add(u1)
session.add(u2)
session.commit()
session.expunge_all()
# --check--
assert not(_get_user() is None)


##############################
## ADD
##############################
_logger.info('-' * 80)
d1 = Document(title=u'Document-1', body=u'Doc1 some body skipped the body')
# when not using SessionExtension for any reason, this can be called manually
#d1.create_activitylog(ACTIVITY_ADD)
session.add(d1)
session.commit()

f1 = Folder(title=u'Folder-1', comment=u'This folder is empty')
# when not using SessionExtension for any reason, this can be called manually
#f1.create_activitylog(ACTIVITY_ADD)
session.add(f1)
session.commit()

# --check--
session.expunge_all()
logs = session.query(ActivityLog).all()
_logger.debug(logs)
assert len(logs) == 2
assert logs[0].activity_type == ACTIVITY_ADD
assert logs[0].target.title == u'Document-1'
assert logs[0].target.title == logs[0].target_title
assert logs[1].activity_type == ACTIVITY_ADD
assert logs[1].target.title == u'Folder-1'
assert logs[1].target.title == logs[1].target_title

##############################
## MOD(ify)
##############################
_logger.info('-' * 80)
session.expunge_all()
d1 = session.query(Document).filter_by(id=1).one()
assert d1.title == u'Document-1'
assert d1.body == u'Doc1 some body skipped the body'
assert d1.activitylogs == []
d1.title = u'Modified: Document-1'
d1.body = u'Modified: body'
# when not using SessionExtension (or it does not work, this can be called manually)
#d1.create_activitylog(ACTIVITY_MOD)
session.commit()
_logger.debug(d1.activitylogs_readonly)

# --check--
session.expunge_all()
logs = session.query(ActivityLog).all()
assert len(logs)==3
assert logs[2].activity_type == ACTIVITY_MOD
assert logs[2].target.title == u'Modified: Document-1'
assert logs[2].target.title == logs[2].target_title


##############################
## DEL(ete)
##############################
_logger.info('-' * 80)
session.expunge_all()
d1 = session.query(Document).filter_by(id=1).one()
# when not using SessionExtension for any reason, this can be called manually,
#d1.create_activitylog(ACTIVITY_DEL)
session.delete(d1)
session.commit()
session.expunge_all()

# --check--
session.expunge_all()
logs = session.query(ActivityLog).all()
assert len(logs)==4
assert logs[0].target is None
assert logs[2].target is None
assert logs[3].activity_type == ACTIVITY_DEL
assert logs[3].target is None

##############################
## print all activity logs
##############################
_logger.info('=' * 80)
logs = session.query(ActivityLog).all()
for log in logs:
    _ = log.target
    _logger.info("%s -> %s", log, log.target)

##############################
## navigate from main object
##############################
_logger.info('=' * 80)
session.expunge_all()
f1 = session.query(Folder).filter_by(id=1).one()
_logger.info(f1.activitylogs_readonly)