Dynamically constructing filters in SQLAlchemy

In case this is useful to someone, here is what I ended up doing:

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

This covers all SQLAlchemy column comparators:

• eq for ==
• lt for <
• ge for >=
• in for in_
• like for like
• etc.

The exhaustive list with their corresponding names can be found here.

One useful trick while building multiple expression filter:

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

Using this approach will allow you to combine expressions with and/or logic. Also this will allow you to avoid recursion calls like in your answer.

You can use sqlalchemy-elasticquery for construct dynamically filters using SQLAlchemy.

?filters={ "age" : 3 }

improving https://stackoverflow.com/a/14876320/12562701, you can get more complex filters

Dict to map op

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

Class dao:

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