在SQLAlchemy中过滤关系属性



我有一些带有Widget对象的代码,必须定期进行一些处理。小工具与Process对象有关系,该对象跟踪各个处理尝试并保存有关这些尝试的数据,如状态信息、开始和结束时间以及结果。这种关系看起来像这样:

class Widget(Base):
_tablename_ = 'widget'
id = Column(Integer, primary_key=True)
name = Column(String)
attempts = relationship('Process')
class Process(Base):
_tablename_ = 'process'
id = Column(Integer, primary_key=True)
widget_id = Column(Integer, ForeignKey='widget.id'))
start = Column(DateTime)
end = Column(DateTime)
success = Column(Boolean)

我想在Widget上有一个方法来检查是否到了处理该小部件的时候。它需要查看所有的尝试,找到最近成功的一次,看看它是否超过了阈值。

一种选择是使用列表理解来迭代Widget.attempts。假设nowdelay是合理的datetimetimedelta对象,那么当定义为Widget:上的方法时,类似的东西就起作用了

def ready(self):
recent_success = [attempt for attempt in self.attempts if attempt.success is True and attempt.end >= now - delay]
if recent_success:
return False
return True

这似乎是一个很好的惯用Python,但它没有很好地利用支持数据的SQL数据库的功能,而且它可能不如运行类似的SQL查询高效,尤其是在attempts列表中有大量Process对象的情况下。不过,我很难找到将其作为查询实现的最佳方式。

Widget内部运行查询很容易,类似于以下内容:

def ready(self):
recent_success = session.query(Process).filter(
and_(
Process.widget_id == self.id,
Process.success == True,
Process.end >= now - delay
)
).order_by(Process.end.desc()).first()
if recent_success:
return False
return True

但是,在单元测试中,我遇到了在定义Widget的模块中正确设置session的问题。在我看来,这是一个糟糕的风格选择,可能不是SQLAlchemy对象的结构化方式。

我可以使ready()函数成为Widget类的外部函数,这将解决在单元测试中设置session的问题,但这似乎是糟糕的OO结构。

我认为理想的情况是,如果我能用类似SQL的代码过滤Widget.attempts,这比列表理解更有效,但我还没有发现任何迹象表明这是可能的。

对于这样的事情,什么是最好的方法?

您正在朝着正确的方向思考。Widget实例中的任何解决方案都意味着您需要处理所有实例。寻求外部过程将具有更好的性能和更容易的可测试性。

您可以使用以下查询获取所有需要安排下一次处理的Widget实例:

q = (
session
.query(Widget)
.filter(Widget.attempts.any(and_(
Process.success == True,
Process.end >= now - delay,
)))
)
widgets_to_process = q.all()

如果你真的想在模型上有一个属性,我不会创建一个单独的查询,而是使用关系:

def ready(self, at_time):
successes = [
attempt 
for attempt in sorted(self.attempts, key=lambda v: v.end)
if attempt.success and attempt.end >= at_time  # at_time = now - delay
]
return bool(successes)

最新更新