Динамическое построение фильтров в SQLAlchemy

Я ищу способ динамически создавать фильтры с помощью SQLAlchemy. То есть, учитывая столбец, имя оператора и значение сравнения, создайте соответствующий фильтр.

Я попытаюсь проиллюстрировать пример (это будет использовано для создания API). Скажем, мы имеем следующую модель:

class Cat(Model):

  id = Column(Integer, primary_key=True)
  name = Column(String)
  age = Column(Integer)

Я хотел бы сопоставлять запросы с фильтрами. Например,

/cats?filter=age;eq;3 должен генерировать Cat.query.filter(Cat.age == 3)
/cats?filter=age;in;5,6,7&filter=id;ge;10 должен генерировать Cat.query.filter(Cat.age.in_([5, 6, 7])).filter(Cat.id >= 10)

Я огляделся, чтобы посмотреть, как это было сделано, но не смог найти способ, который не включал бы вручную сопоставление каждого имени оператора компаратору или что-то подобное. Например, Flask-Restless хранит словарь всех поддерживаемых операций и сохраняет соответствующие лямбда-функции (здесь)).

Я искал в документах SQLAlchemy и нашел два потенциальных вывода, но ни один из них не казался удовлетворительным:

используя Column.like, Column.in_...: эти операторы доступны непосредственно в столбце, который упростит использование getattr, но некоторые из них все еще отсутствуют (==, > и т.д.)..).
используя Column.op: например. Cat.name.op('=')('Hobbes'), но это не работает для всех операторов (in а именно).

Есть ли чистый способ сделать это без lambda функций?

Ответ 1

Если это кому-то полезно, вот что я сделал:

from flask import request

class Parser(object):

  sep = ';'

  # ...

  def filter_query(self, query):
    model_class = self._get_model_class(query) # returns the query Model
    raw_filters = request.args.getlist('filter')
    for raw in raw_filters:
      try:
        key, op, value = raw.split(self.sep, 3)
      except ValueError:
        raise APIError(400, 'Invalid filter: %s' % raw)
      column = getattr(model_class, key, None)
      if not column:
        raise APIError(400, 'Invalid filter column: %s' % key)
      if op == 'in':
        filt = column.in_(value.split(','))
      else:
        try:
          attr = filter(
            lambda e: hasattr(column, e % op),
            ['%s', '%s_', '__%s__']
          )[0] % op
        except IndexError:
          raise APIError(400, 'Invalid filter operator: %s' % op)
        if value == 'null':
          value = None
        filt = getattr(column, attr)(value)
      query = query.filter(filt)
    return query

Это охватывает все компараторы столбцов SQLAlchemy:

eq для ==
lt для <
ge для >=
in для in_
like для like
и др.

Полный список с их соответствующими именами можно найти здесь.

Ответ 2

Один полезный трюк при построении фильтра множественных выражений:

filter_group = list(Column.in_('a','b'),Column.like('%a'))
query = query.filter(and_(*filter_group))

Используя этот подход, вы сможете комбинировать выражения с и/или логикой. Также это позволит вам избежать вызовов рекурсии, например, в вашем ответе.

Ответ 3

Вы можете использовать sqlalchemy-elasticquery для создания динамических фильтров с использованием SQLAlchemy.

?filters={ "age" : 3 }

Ответ 4

class Place(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    search_id = db.Column(db.Integer, db.ForeignKey('search.id'), nullable=False)

    @classmethod
    def dinamic_filter(model_class, filter_condition):
        '''
        Return filtered queryset based on condition.
        :param query: takes query
        :param filter_condition: Its a list, ie: [(key,operator,value)]
        operator list:
            eq for ==
            lt for <
            ge for >=
            in for in_
            like for like
            value could be list or a string
        :return: queryset
        '''
        __query = db.session.query(model_class)
        for raw in filter_condition:
            try:
                key, op, value = raw
            except ValueError:
                raise Exception('Invalid filter: %s' % raw)
            column = getattr(model_class, key, None)
            if not column:
                raise Exception('Invalid filter column: %s' % key)
            if op == 'in':
                if isinstance(value, list):
                    filt = column.in_(value)
                else:
                    filt = column.in_(value.split(','))
            else:
                try:
                    attr = list(filter(lambda e: hasattr(column, e % op), ['%s', '%s_', '__%s__']))[0] % op
                except IndexError:
                    raise Exception('Invalid filter operator: %s' % op)
                if value == 'null':
                    value = None
                filt = getattr(column, attr)(value)
            __query = __query.filter(filt)
        return __query

Выполнить как:

places = Place.dinamic_filter([('search_id', 'eq', 1)]).all()