Фильтрация атрибута отношения в SQLAlchemy

Question

У меня есть код с Widget объектом, который должен go периодически обрабатываться. Виджеты имеют отношение к объекту 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

Это выглядит как хорошая идиоматика c Python, но он не в полной мере использует возможности базы данных SQL для резервного копирования данных и, вероятно, менее эффективен, чем выполнение аналогичного запроса SQL, особенно если в * имеется большое количество объектов Process. 1018 * список. Хотя мне трудно найти лучший способ реализовать это как запрос.

Достаточно просто выполнить запрос внутри 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

Но я сталкиваюсь с проблемами в модульных тестах с правильной настройкой session внутри модуля, который определяет Widget. Мне кажется, это плохой выбор стиля, и, вероятно, не то, как объекты SQLAlchemy должны быть структурированы.

Я мог бы сделать функцию ready() чем-то внешним по отношению к классу Widget, что решило бы проблемы с установкой session в модульных тестах, но это похоже на плохую структуру OO.

I думаю, что идеальным было бы, если бы я мог каким-то образом фильтровать Widget.attempts с SQL -подобным кодом, который более эффективен, чем понимание списка, но я не нашел ничего, что бы предполагало, что это возможно.

Что на самом деле лучший подход для чего-то вроде этого?

Answer 1

Вы думаете в правильном направлении. Любое решение в экземпляре Widget подразумевает необходимость обработки всех экземпляров. Поиск процесса external будет иметь более высокую производительность и более легкую тестируемость.

Вы можете получить все экземпляры 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)