Как я могу сопоставить иерархию в SQLAlchemy с динамическими отношениями?

Question

Я определяю модель SQLAlchemy следующим образом:

class SubProject(Base):
  active = Column(Boolean)

class Project(Base):
  active = Column(Boolean)
  subprojects = relationship(SubProject, backref=backref('project'))

class Customer(Base):
  active = Column(Boolean)
  projects = relationship(Project, backref=backref('customer'))

Мне нужно получить список клиентов в одном из следующих двух состояний:

1. Все клиенты со всеми проектами и всеми подпроектами

2. Только активные клиенты, только с активными проектами и только активными подпроектами

   Редактировать Примечательно, что все активные клиенты, у которых нет проектов, должны быть включены, и все активные проекты, которые не имеют активных опросов, должны быть включены в это.

Это было бы тривиально в SQL с объединением, но я не знаю, как это сделать с помощью SQLAlchemy ORM. Какое здесь решение?

Answer 1

Чтобы дополнить ответ Дениса, вы можете использовать enable_assertions (False). Из того, что я понял, это дополнительная проверка запросов, которые SQLAlchemy добавляет для обычных операций. Для более сложных ситуаций вы можете отключить его.

Answer 2

Если я вас правильно понял, вам нужно, чтобы все неактивные объекты стали невидимыми для ваших запросов.Следующий класс отфильтрует все объекты модели с атрибутом active, установленным на False, включая те, к которым осуществляется доступ через отношения:

from sqlalchemy.orm import Query
from sqlalchemy.orm.util import _class_to_mapper


class QueryActive(Query):

    def __init__(self, entities, *args, **kwargs):
        Query.__init__(self, entities, *args, **kwargs)
        query = self
        for entity in entities:
            if hasattr(entity, 'parententity'):
                entity = entity.parententity
            cls = _class_to_mapper(entity).class_
            if hasattr(cls, 'active'):
                query = query.filter(cls.active==True)
        self._criterion = query._criterion

    def get(self, ident):
        # Use default implementation when there is no condition
        if not self._criterion:
            return Query.get(self, ident)
        # Copied from Query implementation with some changes.
        if hasattr(ident, '__composite_values__'):
            ident = ident.__composite_values__()
        mapper = self._only_mapper_zero(
                    "get() can only be used against a single mapped class.")
        key = mapper.identity_key_from_primary_key(ident)
        if ident is None:
            if key is not None:
                ident = key[1]
        else:
            from sqlalchemy import util
            ident = util.to_list(ident)
        if ident is not None:
            columns = list(mapper.primary_key)
            if len(columns)!=len(ident):
                raise TypeError("Number of values doesn't match number "
                                'of columns in primary key')
            params = {}
            for column, value in zip(columns, ident):
                params[column.key] = value
            return self.filter_by(**params).first()

Чтобы использовать его, необходимо создать отдельный объект сеанса:

session_active = sessionmaker(bind=engine, query_cls=QueryActive)()

Такой подход имеет ограничения и не подходит для некоторых сложных запросов, но подходит для большинства проектов.