sqlalchemy bind values

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 retri开发者_JAVA百科eve 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')]