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.

Leave a Comment

Hata!: SQLSTATE[HY000] [1045] Access denied for user 'divattrend_liink'@'localhost' (using password: YES)