I want to get the value of a parameter in a sqlalchemy query object dynamically:
q = session.query(Model).filter(Model.foo = 6)
I now want to be able to retrieve the value 6 from q
assert(q.magic == 6)
Attempts:
print(q._criterion) # -> models.foo = :foo_1
But where is the value of foo_1?
SQLAlchemy generates a tree structure from your filter predicates, appending each leaf on as appropriate and putting the result in Query._criterion. You can explore this with the get_children() method of various ClauseElement and ColumnElement classes.
For Model.foo == 6 you'll end up with something like this:
Model.foo == 6
|
_BinaryExpression
/ \
/ \
/ \
Column('foo', Integer(), _BindParamClause(u'%(168261004 foo)s',
...) 6, type_=Integer())
If you were to & together two predicates, like (Model.foo == 6) & (Model.name == 'a name') or by chaining filter calls, you'd get a BooleanClauseList with two _BinaryExpression children. This means you can't hard code a simple expression to reliably return the values you want but instead have to traverse the predicate tree.
The traverse function from sqlalchemy.sql.visitors does just that, relying on a dictionary of special names that relate each Element's __visit_name__ attribute to a processing function. This is a breadth-first traversal, which as you can see is appropriate for the example below; there's also a depth-first version available.
The following function shows how to generate a list of column-parameter pairs from a given query. I adapted it from the Beaker caching example:
def extract_cols_params(query):
if query._criterion is None:
return []
c, v = [], []
def visit_bindparam(bind):
value = query._params.get(bind.key, bind.value)
if callable(value):
value = value()
v.append(value)
def visit_column(col):
c.append('%s.%s' % (col.table.name, col.name))
visitors.traverse(query._criterion, # our predicate tree
{}, # kwargs for the iterator used by
# the traversal; undeeded.
{'bindparam': visit_bindparam, # for _BindParamClauses
'column' : visit_column}) # for Columns
return zip(c, v)
>>> extract_cols_params(Session.query(Model).filter((Model.foo == 6)
).filter(Model.name == 'a name'))
[('models.foo', 6), ('models.name', 'a name')]
