Каков наилучший способ предварительно фильтровать доступ пользователей для запросов sqlalchemy?

Question

Я смотрел рецепты sqlalchemy на их вики, но не знаю, какой из них лучше всего реализовать то, что я пытаюсь сделать.

С каждой строкой в ​​моих таблицах связан идентификатор_пользователя. Прямо сейчас, для каждого запроса я запрашиваю идентификатор пользователя, который в данный момент вошел в систему, а затем запрашиваю по интересующим меня критериям. Я обеспокоен тем, что разработчики могут забыть добавить этот фильтр в запрос (огромный риск для безопасности ). Поэтому я хотел бы установить глобальный фильтр на основе прав администратора текущего пользователя, чтобы фильтровать то, что мог видеть вошедший в систему пользователь.

Ценю вашу помощь. Спасибо.

Answer 1

Ниже приведен упрощенный переопределенный конструктор запросов для фильтрации всех запросов модели (включая отношения). Вы можете передать его как query_cls параметр sessionmaker. Параметр ID пользователя не должен быть глобальным, поскольку сессия создается, когда она уже доступна.

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

Он работает только для запросов одной модели, и есть много работы, чтобы сделать его пригодным для других случаев. Я хотел бы увидеть детализированную версию, поскольку она ДОЛЖНА ИМЕТЬ функциональность для большинства веб-приложений. Он использует фиксированные условия, хранящиеся в каждом классе модели, поэтому вам нужно изменить его в соответствии с вашими потребностями.

Answer 2

Я написал расширение SQLAlchemy, которое, я думаю, выполняет то, что вы описываете: https://github.com/mwhite/multialchemy

Это достигается путем передачи изменений свойств Query._from_obj и QueryContext._froms, в которых в конечном итоге устанавливаются таблицы для выбора.

Answer 3

Вот очень наивная реализация, которая предполагает наличие атрибута / свойства self.current_user, вошедшего в систему пользователя, сохраненного.

class YourBaseRequestHandler(object):

    @property
    def current_user(self):
        """The current user logged in."""
        pass

    def query(self, session, entities):
        """Use this method instead of :method:`Session.query()
        <sqlalchemy.orm.session.Session.query>`.

        """
        return session.query(entities).filter_by(user_id=self.current_user.id)