значения привязки sqlalchemy

Question

Я хочу получить значение параметра в объекте запроса sqlalchemy динамически:

q = session.query(Model).filter(Model.foo = 6)

Теперь я хочу иметь возможность получить значение 6 из q

assert(q.magic == 6)

Попытки: напечатать q._criterion # -> models.foo =: ​​foo_1 Но где же значение foo_1?

Answer 1

SQLAlchemy генерирует древовидную структуру из ваших предикатов фильтра, добавляя каждый лист соответствующим образом и помещая результат в Query._criterion. Вы можете исследовать это с помощью метода get_children() различных классов ClauseElement и ColumnElement.

Для Model.foo == 6 вы получите что-то вроде этого:

                    Model.foo == 6        
                           |                 
                   _BinaryExpression
                          / \
                         /   \
                        /     \   
Column('foo', Integer(),      _BindParamClause(u'%(168261004 foo)s',
              ...)                             6, type_=Integer())  

Если бы вы были вместе с двумя предикатами, такими как (Model.foo == 6) & (Model.name == 'a name') или с помощью цепочки вызовов filter, вы бы получили BooleanClauseList с двумя _BinaryExpression детьми. Это означает, что вы не можете жестко запрограммировать простое выражение, чтобы надежно возвращать нужные значения, но вместо этого придется обходить дерево предикатов.

Функция traverse из sqlalchemy.sql.visitors делает именно это, полагаясь на словарь специальных имен, которые связывают атрибут __visit_name__ каждого элемента с функцией обработки. Это обход в ширину, который, как вы можете видеть, подходит для примера ниже; также доступна первая версия глубины.

Следующая функция показывает, как сгенерировать список пар столбец-параметр из заданного запроса. Я адаптировал это из Пример кэширования Beaker :

 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')]