How can I map a hierarchy in SQLAlchemy with dynamic relationships?

Question

I am defining a SQLAlchemy model like this:

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'))

I need to get the list of customers in one of these two states:

1. All customers, with all projects and all sub projects

2. Only active customers, with only active projects, and only active subprojects

   Edit Notably, all active customers that have no projects should be included, and all active projects that have no active surveys should be included in this.

This would be trivial in SQL with a join, but I'm at a loss as to how to accomplish it using SQLAlchemy ORM. What's the solution here?

Answer 1

If I understand you right, you need that all inactive objects become invisible to your queries. The following class will filter out all model objects with active attribute set to False, including those accessed through relations:

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()

To use it you have to create a separate session object:

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

Such approach has limitations and won't for some complex queries, but is OK for most projects.