Как получить необработанный скомпилированный SQL-запрос из выражения SQLAlchemy?

У меня есть объект запроса SQLAlchemy, и я хочу получить текст скомпилированного оператора SQL со всеми его параметрами (например, никакие %sили другие переменные, ожидающие связывания компилятором операторов или механизмом диалекта MySQLdb и т. Д.).

Вызов str()запроса показывает что-то вроде этого:

SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC

Я пробовал искать в query._params, но это пустой dict. Я написал свой собственный компилятор, используя этот пример sqlalchemy.ext.compiler.compilesдекоратора, но даже там есть оператор, %sгде мне нужны данные.

Я не могу понять, когда мои параметры смешиваются для создания запроса; при изучении объекта запроса они всегда являются пустым словарем (хотя запрос выполняется нормально, и механизм его распечатывает, когда вы включаете регистрацию эха).

Я начинаю получать сообщение о том, что SQLAlchemy не хочет, чтобы я знал базовый запрос, поскольку он нарушает общий характер интерфейса API выражений для всех различных API-интерфейсов DB. Я не возражаю, если запрос будет выполнен до того, как я узнаю, что это было; Я просто хочу знать!

Этот блог предоставляет обновленный ответ.

Цитируя сообщение в блоге, это предложено и сработало для меня.

>>> from sqlalchemy.dialects import postgresql
>>> print str(q.statement.compile(dialect=postgresql.dialect()))

Где q определяется как:

>>> q = DBSession.query(model.Name).distinct(model.Name.value) \
             .order_by(model.Name.value)

Или просто любой тип session.query ().

Спасибо Николя Каду за ответ! Я надеюсь, что это поможет другим, кто ищет сюда.

В документации используется literal_bindsдля печати запроса с qпараметрами:

print(q.statement.compile(compile_kwargs={"literal_binds": True}))

У вышеупомянутого подхода есть предостережения, что он поддерживается только для базовых типов, таких как целые числа и строки, и, кроме того, если bindparam () без предварительно установленного значения используется напрямую, он также не сможет преобразовать его в строку.

В документации также содержится это предупреждение:

Никогда не используйте этот метод со строковым содержимым, полученным из ненадежных входных данных, например из веб-форм или других приложений пользовательского ввода. Средства SQLAlchemy для преобразования значений Python в прямые строковые значения SQL не защищены от ненадежного ввода и не проверяют тип передаваемых данных. Всегда используйте связанные параметры при программном вызове операторов SQL, отличных от DDL, для реляционной базы данных.

Это должно работать с Sqlalchemy> = 0.6

from sqlalchemy.sql import compiler

from psycopg2.extensions import adapt as sqlescape
# or use the appropiate escape function from your db driver

def compile_query(query):
    dialect = query.session.bind.dialect
    statement = query.statement
    comp = compiler.SQLCompiler(dialect, statement)
    comp.compile()
    enc = dialect.encoding
    params = {}
    for k,v in comp.params.iteritems():
        if isinstance(v, unicode):
            v = v.encode(enc)
        params[k] = sqlescape(v)
    return (comp.string.encode(enc) % params).decode(enc)

Для бэкэнда MySQLdb я немного изменил потрясающий ответ Альбертова (большое спасибо!). Я уверен, что их можно объединить, чтобы проверить, было ли comp.positionalэто, Trueно это немного выходит за рамки этого вопроса.

def compile_query(query):
    from sqlalchemy.sql import compiler
    from MySQLdb.converters import conversions, escape

    dialect = query.session.bind.dialect
    statement = query.statement
    comp = compiler.SQLCompiler(dialect, statement)
    comp.compile()
    enc = dialect.encoding
    params = []
    for k in comp.positiontup:
        v = comp.params[k]
        if isinstance(v, unicode):
            v = v.encode(enc)
        params.append( escape(v, conversions) )
    return (comp.string.encode(enc) % tuple(params)).decode(enc)

Дело в том, что sqlalchemy никогда не смешивает данные с вашим запросом. Запрос и данные передаются отдельно в ваш базовый драйвер базы данных - интерполяция данных происходит в вашей базе данных.

Sqlalchemy передает запрос str(myquery)в базу данных, как вы видели , и значения будут помещены в отдельный кортеж.

Вы можете использовать какой-то подход, в котором вы сами интерполируете данные с запросом (как предлагает Альбертов ниже), но это не то же самое, что выполняет sqlalchemy.

Во-первых, позвольте мне предисловие, сказав, что я предполагаю, что вы делаете это в основном для целей отладки - я бы не рекомендовал пытаться изменить оператор вне SQLAlchemy fluent API.

К сожалению, не существует простого способа показать скомпилированный оператор с включенными параметрами запроса. SQLAlchemy фактически не помещает параметры в инструкцию - они передаются в ядро ​​базы данных в виде словаря . Это позволяет библиотеке для конкретной базы данных обрабатывать такие вещи, как экранирование специальных символов, чтобы избежать внедрения SQL.

Но вы можете довольно легко сделать это в два этапа. Чтобы получить инструкцию, вы можете сделать, как вы уже показали, и просто распечатать запрос:

>>> print(query)
SELECT field_1, field_2 FROM table WHERE id=%s;

Вы можете приблизиться на один шаг с помощью query.statement, чтобы увидеть имена параметров. Обратите внимание: :id_1ниже и %sвыше - это не проблема в этом очень простом примере, но может быть ключевым в более сложном заявлении.

>>> print(query.statement)
>>> print(query.statement.compile()) # seems to be equivalent, you can also
                                     # pass in a dialect if you want
SELECT field_1, field_2 FROM table WHERE id=:id_1;

Затем вы можете получить фактические значения параметров, получив paramsсвойство скомпилированного оператора:

>>> print(query.statement.compile().params)
{u'id_1': 1} 

По крайней мере, это сработало для серверной части MySQL; Я ожидал, что он также достаточно общий для PostgreSQL, и его не нужно использовать psycopg2.

Для серверной части postgresql, использующей psycopg2, вы можете прослушивать do_executeсобытие, а затем использовать курсор, оператор и параметры с принудительным вводом вместе с Cursor.mogrify()для встраивания параметров. Вы можете вернуть True, чтобы предотвратить фактическое выполнение запроса.

import sqlalchemy

class QueryDebugger(object):
    def __init__(self, engine, query):
        with engine.connect() as connection:
            try:
                sqlalchemy.event.listen(engine, "do_execute", self.receive_do_execute)
                connection.execute(query)
            finally:
                sqlalchemy.event.remove(engine, "do_execute", self.receive_do_execute)

    def receive_do_execute(self, cursor, statement, parameters, context):
        self.statement = statement
        self.parameters = parameters
        self.query = cursor.mogrify(statement, parameters)
        # Don't actually execute
        return True

Пример использования:

>>> engine = sqlalchemy.create_engine("postgresql://postgres@localhost/test")
>>> metadata = sqlalchemy.MetaData()
>>> users = sqlalchemy.Table('users', metadata, sqlalchemy.Column("_id", sqlalchemy.String, primary_key=True), sqlalchemy.Column("document", sqlalchemy.dialects.postgresql.JSONB))
>>> s = sqlalchemy.select([users.c.document.label("foobar")]).where(users.c.document.contains({"profile": {"iid": "something"}}))
>>> q = QueryDebugger(engine, s)
>>> q.query
'SELECT users.document AS foobar \nFROM users \nWHERE users.document @> \'{"profile": {"iid": "something"}}\''
>>> q.statement
'SELECT users.document AS foobar \nFROM users \nWHERE users.document @> %(document_1)s'
>>> q.parameters
{'document_1': '{"profile": {"iid": "something"}}'}

Следующее решение использует язык выражений SQLAlchemy и работает с SQLAlchemy 1.1. Это решение не смешивает параметры с запросом (как было запрошено исходным автором), но предоставляет способ использования моделей SQLAlchemy для создания строк запроса SQL и словарей параметров для различных диалектов SQL. Пример основан на учебнике http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html.

Учитывая класс,

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class foo(Base):
    __tablename__ = 'foo'
    id = Column(Integer(), primary_key=True)
    name = Column(String(80), unique=True)
    value = Column(Integer())

мы можем создать оператор запроса, используя функцию выбора .

from sqlalchemy.sql import select    
statement = select([foo.name, foo.value]).where(foo.value > 0)

Затем мы можем скомпилировать инструкцию в объект запроса.

query = statement.compile()

По умолчанию оператор компилируется с использованием базовой «именованной» реализации, совместимой с базами данных SQL, такими как SQLite и Oracle. Если вам нужно указать диалект, например PostgreSQL, вы можете сделать

from sqlalchemy.dialects import postgresql
query = statement.compile(dialect=postgresql.dialect())

Или, если вы хотите явно указать диалект как SQLite, вы можете изменить стиль параметра с «qmark» на «named».

from sqlalchemy.dialects import sqlite
query = statement.compile(dialect=sqlite.dialect(paramstyle="named"))

Из объекта запроса мы можем извлечь строку запроса и параметры запроса.

query_str = str(query)
query_params = query.params

и, наконец, выполните запрос.

conn.execute( query_str, query_params )

Вы можете использовать события из семейства ConnectionEvents : after_cursor_executeилиbefore_cursor_execute .

В sqlalchemy UsageRecipes от @zzzeek вы можете найти этот пример:

Profiling

...
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement,
                        parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())
    logger.debug("Start Query: %s" % statement % parameters)
...

Здесь вы можете получить доступ к своей выписке

Итак, собрав множество маленьких кусочков этих разных ответов, я придумал то, что мне было нужно: простой набор кода, который нужно было вставлять и время от времени, но надежно (то есть обрабатывать все типы данных) получать точный скомпилированный SQL, отправленный на мой адрес. Серверная часть Postgres, просто запросив сам запрос:

from sqlalchemy.dialects import postgresql

query = [ .... some ORM query .... ]

compiled_query = query.statement.compile(
    dialect=postgresql.dialect()
)
mogrified_query = session.connection().connection.cursor().mogrify(
    str(compiled_query),
    compiled_query.params
)

print("compiled SQL = {s}".format(mogrified_query.decode())

Я думаю, что .statement, возможно, поможет: http://docs.sqlalchemy.org/en/latest/orm/query.html?highlight=query

>>> local_session.query(sqlalchemy_declarative.SomeTable.text).statement
<sqlalchemy.sql.annotation.AnnotatedSelect at 0x6c75a20; AnnotatedSelectobject>
>>> x=local_session.query(sqlalchemy_declarative.SomeTable.text).statement
>>> print(x)
SELECT sometable.text 
FROM sometable

Я создал эту небольшую функцию, которую импортирую, когда хочу распечатать полный запрос, учитывая, что я нахожусь в середине теста, когда диалект уже привязан:

import re

def print_query(query):
    regex = re.compile(":(?P<name>\w+)")
    params = query.statement.compile().params
    sql = regex.sub("'{\g<name>}'", str(query.statement)).format(**params)
    print(f"\nPrinting SQLAlchemy query:\n\n")
    print(sql)
    return sql