Я смотрел на sqlalchemy рецепты на их Wiki, но не знаю, какой является лучшим для реализации то, что я пытаюсь сделать.
Каждая строка на в моих таблицах имеет user_id, связанный с ним. Прямо сейчас, для каждого запроса, я запросил идентификатором пользователя, это в настоящее время входило в систему, затем запрашивает по критериям, которыми я интересуюсь. Мое беспокойство - то, что разработчики могли бы забыть добавлять этот фильтр к запросу (огромная угроза безопасности). Поэтому я хотел бы установить глобальный фильтр на основе прав администратора текущего пользователя для фильтрации то, что видел зарегистрированный пользователь.
Цените свою справку.Спасибо.
Ниже приведен упрощенный переопределенный конструктор запросов для фильтрации всех запросов модели (включая отношения). Вы можете передать его как параметр query_cls
в sessionmaker
. Параметр идентификатора пользователя не обязательно должен быть глобальным, поскольку сеанс создается, когда он уже доступен.
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 directly related to the problem, but is useful too.
query = HackedQuery(entities, session).with_polymorphic('*')
# Version for several entities needs thorough testing, so we
# don't use it yet.
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
Он работает только для запросов одной модели, и предстоит много работы, чтобы сделать его пригодным для других случаев. Я бы хотел увидеть доработанную версию, поскольку она ДОЛЖНА ИМЕТЬ функциональность для большинства веб-приложений. Он использует фиксированное условие, хранящееся в каждом классе модели, поэтому вы должны изменить его в соответствии с вашими потребностями.