I am looking for a way to dynamically construct filters using SQLAlchemy. That is, given the column, the operator name and the comparing value, construct the corresponding filter.
I'll try to illustrate using an example (this would be used to build an API). Let's say we have the following model:
class Cat(Model):
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
I would like to map queries to filters. For example,
/cats?filter=age;eq;3
should generate Cat.query.filter(Cat.age == 3)
/cats?filter=age;in;5,6,7&filter=id;ge;10
should generate Cat.query.filter(Cat.age.in_([5, 6, 7])).filter(Cat.id >= 10)
I looked around to see how it has been done but couldn't find a way that didn't involve manually mapping each operator name to a comparator or something similar. For instance, Flask-Restless keeps a dictionary of all supported operations and stores the corresponding lambda functions (code here).
I searched in the SQLAlchemy docs and found two potential leads but neither seemed satisfying:
using Column.like
, Column.in_
...: these operators are available directly on the column which would make it simple using getattr
but some are still missing (==
, >
, etc.).
using Column.op
: e.g. Cat.name.op('=')('Hobbes')
but this doesn't seem to work for all operators (in
namely).
Is there a clean way to do this without lambda
functions?
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
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.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With