使用SQLAlchemy ORM，如何监听使用"noload"或"raise"来设置延迟加载的关系调用？

Question

我的示例表：

class TestEvent(Model):
    id = Column(Integer, primary_key=True)
    name = Column(String(20))

    def __str__(self):
        return f'{self.name} ({self.id})'


class TestMarket(Model):
    id = Column(Integer, primary_key=True)
    name = Column(String(20))

    def __str__(self):
        return f'{self.name} ({self.id})'


class TestEventMarket(Model):
    event_id = Column(Integer, ForeignKey(TestEvent.id))
    market_id = Column(Integer, ForeignKey(TestMarket.id))

    event = relationship(TestEvent, lazy='noload',
                         backref=backref('event_markets', lazy='dynamic'))
    market = relationship(TestMarket, lazy='noload',
                          backref=backref('event_markets', lazy='dynamic'))

    PrimaryKeyConstraint(event_id, market_id)

    def __str__(self):
        return f'{self.event_id}:{self.market_id}'

初始化代码示例：

event = TestEvent(name='test event1')
market = TestMarket(name='test market1')

em = TestEventMarket(event=event, market=market)

session.add(event)
session.add(market)
session.add(em)
session.commit()

示例查询代码：

event = TestEvent.query.first()
market = TestMarket.query.first()
em = TestEventMarket.query.first()

print(event, market, em)
print(list(map(str, event.event_markets)),
      list(map(str, market.event_markets)),
      em.event, em.market)

预期输出：

test event1 (1) test market1 (1) 1:1
['1:1'] ['1:1'] None None

我想要这种行为的原因是，在我的应用程序中，在启动时，我获取所有“相关的”TestEvent和TestMarket，并缓存它们。一个新的TestEvent或TestMarket可能会在生命周期的某个时刻创建，但这只是添加到缓存中，所以一切正常。你知道吗

这些是示例表，但我还有一些基本上作为枚举表运行的其他表，它们在整个应用程序中使用。因为我缓存了这些数据，并且有一种“标准”的方法来访问它们，所以我更希望数据库模型使用这种方法。你知道吗

我浏览了SQLAlchemy文档，认为可以用它们的MappedCollection，或者它们的HybridAttributes来完成，但是作为它们对我来说完全是新的概念，我很难理解它们，弄清楚它们是否正是我要找的。或者我应该创建一个自定义的InstrumentedAttribute并拥有自己的relationship？再说一次，它“感觉”不对，我认为应该有更好的方法来实现我想要的。你知道吗

我想覆盖或捕获的main是“get”部分。我希望它基于我自己的应用程序逻辑返回一些东西，而不是什么都不返回，或者一个SQL查询。你知道吗

Answer 1

有一种方法确实“有效”，但我认为它非常难看，而且杀伤力过大，就是在TestEventMarket内部声明的relationship，以及为调用设置的属性

class TestEventMarket(Model):
    event_id = Column(Integer, ForeignKey(TestEvent.id))
    market_id = Column(Integer, ForeignKey(TestMarket.id))

    _event = relationship(TestEvent, lazy='noload',
                          backref=backref('event_markets', lazy='dynamic'))

    @property
    def event(self):
        if self._event:
            return self._event
        return get_event(self.event_id)

    @event.setter
    def event(self, event: TestEvent):
        if event.id:
            self.event_id = event.id
        self._event = event

    _market = relationship(TestMarket, lazy='noload',
                           backref=backref('event_markets', lazy='dynamic'))

    @property
    def market(self):
        if self._market:
            return self._market
        return get_market(self.market_id)

    @market.setter
    def market(self, market: TestMarket):
        if market.id:
            self.market_id = market.id
        self._market = market

    PrimaryKeyConstraint(event_id, market_id)

    def __str__(self):
        return f'{self.event_id}:{self.market_id}'