python - 在 SQLAlchemy 中动态构造过滤器

Question

我正在寻找一种使用 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保留所有支持的操作的字典并存储相应的 lambda 函数（代码here）。

我在 SQLAlchemy 文档中进行了搜索，发现了两个潜在的线索，但似乎都不令人满意：

• using Column.like, Column.in_...：这些运算符可直接在列上使用，这将使使用变得简单，getattr但仍然缺少一些（==、、>等）。

• using Column.op: egCat.name.op('=')('Hobbes')但这似乎不适用于所有运算符（in即）。

有没有一种干净的方法可以在没有功能的情况下做到这一点lambda？

Answer 1

如果这对某人有用，这就是我最终要做的事情：

from flask import request

class Parser(object):

  sep = ';'

  # ...

  def filter_query(self, query):
    model_class = self._get_model_class(query) # returns the query's 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
• 等等

可在此处找到包含相应名称的详尽列表。

Answer 2

构建多个表达式过滤器时的一个有用技巧：

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

使用这种方法将允许您将表达式与和/或逻辑结合起来。此外，这将使您避免像答案中那样的递归调用。

Answer 3

您可以使用sqlalchemy-elasticquery使用 SQLAlchemy 构建动态过滤器。

?filters={ "age" : 3 }

Answer 4

改进https://stackoverflow.com/a/14876320/12562701，可以获得更复杂的过滤器

字典映射操作

dict_filtros_op = {
    '==':'eq',
    '!=':'ne',
    '>':'gt',
    '<':'lt',
    '>=':'ge',
    '<=':'le',
    'like':'like',
    'ilike':'ilike',
    'in':'in'
}

类道：

class BaseDao():
    @classmethod
    @init_db_connection
    def create_query_select(cls, model, filters=None, columns=None):
        return cls.db_session.query(*cls.create_query_columns(model=model, columns=columns))\
            .filter(*cls.create_query_filter(model=model, filters=filters))

    @classmethod
    def create_query_filter(cls, model, filters):
        '''
        return sqlalchemy filter list
        Args:
            model:sqlalchemy  model (classe das tabelas)
            filters: filter dict
                     ex:
                        filters = {
                            'or_1':{
                                'and_1':[('id', '>', 5),('id', '!=', 3)],
                                'and_2':[('fase', '==', 'arquivado')]
                            },
                            'and':[('test', '==', 'test')]
                        }
        Returns:
            filt: sqlalchemy filter list
         '''
        if not filters:
            return []

        filt = []
        for condition in filters:
            if type(filters[condition]) == dict:
                if 'and' in condition:
                    filt.append(and_(*cls.create_query_filter(model, filters[condition])))
                elif 'or' in condition:
                    filt.append(or_(*cls.create_query_filter(model, filters[condition])))
                else:
                    raise Exception('Invalid filter condition: %s' % condition)
                continue
            filt_aux = []
            for t_filter in filters[condition]:
                try:
                    column_name, op, value = t_filter
                except ValueError:
                    raise Exception('Invalid filter: %s' % t_filter)
                if not op in dict_filtros_op:
                    raise Exception('Invalid filter operation: %s' % op)
                column = getattr(model, column_name, None)
                if not column:
                    raise Exception('Invalid filter column: %s' % column_name)
                if dict_filtros_op[op] == 'in':
                    filt.append(column.in_(value))
                else:
                    try:
                        attr = list(filter(lambda e: hasattr(column, e % dict_filtros_op[op]), ['%s', '%s_', '__%s__']))[0] % dict_filtros_op[op]
                    except IndexError:
                        raise Exception('Invalid filter operator: %s' % dict_filtros_op[op])
                    if value == 'null':
                        value = None
                    filt_aux.append(getattr(column, attr)(value))
            if 'and' in condition:
                filt.append(and_(*filt_aux))
            elif 'or' in condition:
                filt.append(or_(*filt_aux))
            else:
                raise Exception('Invalid filter condition: %s' % condition)
        return filt

    @classmethod
    def create_query_columns(cls, model, columns):
        '''
        Return a list of attributes (columns) from the class model
        Args:
            model: sqlalchemy model
            columns: string list
                     ex: ['id', 'cnj']
        Returns:
            cols: list of attributes from the class model
         '''
        if not columns:
            return [model]

        cols = []
        for column in columns:
            attr = getattr(model, column, None)
            if not attr:
                raise Exception('Invalid column name %s' % column)
            cols.append(attr)
        return cols

Answer 5

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()