How can I map a hierarchy in SQLAlchemy with dynamic relationships?
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:
- All customers, with all projects and all sub projects
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 n开发者_JAVA百科o 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?
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.
To augment Denis' answer, you can use enable_assertions(False). From what I gather, this is an extra check on the queries that SQLAlchemy adds for normal operations. For more complicated situations you can disabled it.
精彩评论