我有一些带有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
。假设now
和delay
是合理的datetime
和timedelta
对象,那么当定义为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)