В CMS в таблице для (почти) каждой сущности обычно добавляется поле-флаг, определяющее, должна ли данная сущность показываться на сайте. В коде сайта, соответвенно, необходимо не забывать добавлять соответствующее условие в каждый запрос. При использовании ORM мы автоматичеси получаем связанные сущности, для которых запрос генерируется автоматически. Это удобно, но теперь нам ещё нужно проверять, нужно ли показывать каждый из связанных объектов. Есть ещё множество ситуаций, когда такие проверки или добавление дополнительных условий также необходимы. Шансы, что в большом проекте где-то об этом забудут, близки к 100%. Поэтому очень желательно процесс фильтрации непубличных данных автоматизировать. В django для этих целей используют специально написанный менеджер. В древней библиотеки QPS с некоторым подобием ORM сделано даже лучше: можно для разных тегов выборки определить разные правила формирования запроса и даже правила переноса тегов на связанные обекты.
Как же быть с решением проблемы автоматической фильтрации в SQLAlchemy? Существует возможность при создании сессии подставить свой конструктор запроса через атрибут
query_cls в sessionmaker. Но предлагаемые в архивах Google-группы sqlalchemy решения уже не работают, так как метод Query.get() теперь не предназначен для объектов с условием. Я написал свою реализацию метода, без этого ограничения. Вот что получилось в результате:
class HackedQuery(Query):
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 doen'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()
def QueryPublic(entities, session=None):
# It's not derectly related to the problem, but is useful too.
query = HackedQuery(entities, session).with_polymorphic('*')
# I haven't ever seen examples with several entities, so I can test
# this case.
assert len(entities)==1, entities
cls = _class_to_mapper(entities[0]).class_
public_condition = getattr(cls, 'public_condition', None)
if public_condition is not None:
query = query.filter(public_condition)
return query
