SQLAlchemy: вывести фактический запрос

165

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

Кто-нибудь решил эту проблему в общем виде?

bukzor
источник
1
У меня нет, но вы, вероятно, могли бы создать менее хрупкое решение, нажав на sqlalchemy.engineжурнал SQLAlchemy . Он регистрирует запросы и параметры привязки, вам нужно всего лишь заменить заполнители привязки значениями в готовой строке запроса SQL.
Симон
@Simon: есть две проблемы , связанные с использованием регистратора: 1) он печатает только когда оператор выполняет 2) Я все равно придется сделать строку замены, за исключением того, в этом случае, я бы не знаю точно строку связывать-шаблона и мне придется каким-то образом разобрать его в тексте запроса, что сделает решение более хрупким.
Букзор
Новый URL выглядит как docs.sqlalchemy.org/en/latest/faq/… для @ zzzeek.
Джим DeLaHunt

Ответы:

168

В подавляющем большинстве случаев «строковое преобразование» оператора или запроса SQLAlchemy так же просто, как:

print str(statement)

Это относится как к ORM, Queryтак и к любому select()или другому утверждению.

Примечание : следующий подробный ответ сохраняется в документации по sqlalchemy .

Чтобы получить инструкцию как скомпилированную для определенного диалекта или движка, если сама инструкция еще не связана с какой-либо из них, вы можете передать это в compile () :

print statement.compile(someengine)

или без двигателя:

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

Когда нам передается Queryобъект ORM , для того, чтобы получить compile()доступ к методу, нам нужен только доступ к аксессору .statement :

statement = query.statement
print statement.compile(someengine)

Что касается первоначального условия, что связанные параметры должны быть «встроены» в окончательную строку, то здесь проблема состоит в том, что SQLAlchemy обычно не выполняет эту задачу, поскольку это обрабатывается соответствующим образом DBAPI Python, не говоря уже о том, что обход связанных параметров вероятно, наиболее широко используемые дыры в безопасности в современных веб-приложениях. SQLAlchemy имеет ограниченные возможности для выполнения этой последовательности при определенных обстоятельствах, таких как испускание DDL. Чтобы получить доступ к этой функции, можно использовать флаг literal_binds, передаваемый compile_kwargs:

from sqlalchemy.sql import table, column, select

t = table('t', column('x'))

s = select([t]).where(t.c.x == 5)

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

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

Для поддержки встроенного рендеринга литералов для типов, которые не поддерживаются, реализуйте TypeDecoratorдля целевого типа, который включает TypeDecorator.process_literal_paramметод:

from sqlalchemy import TypeDecorator, Integer


class MyFancyType(TypeDecorator):
    impl = Integer

    def process_literal_param(self, value, dialect):
        return "my_fancy_formatting(%s)" % value

from sqlalchemy import Table, Column, MetaData

tab = Table('mytable', MetaData(), Column('x', MyFancyType()))

print(
    tab.select().where(tab.c.x > 5).compile(
        compile_kwargs={"literal_binds": True})
)

производя продукцию как:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)
zzzeek
источник
2
Это не помещает кавычки вокруг строк и не разрешает некоторые связанные параметры.
Букзор
1
вторая половина ответа была обновлена ​​с последней информацией.
zzzeek
2
@zzzeek Почему запросы красивой печати не включены в sqlalchemy по умолчанию? Как query.prettyprint(). Это облегчает боль отладки с большими запросами безмерно.
Jmagnusson
2
@jmagnusson, потому что красота - в глазах смотрящего :) Существует множество хуков (например, событие cursor_execute, фильтры журналов Python @compilesи т. д.) для любого количества сторонних пакетов для реализации систем симпатичной печати.
zzzeek
1
@buzkor re: ограничение, которое было исправлено в 1.0 bitbucket.org/zzzeek/sqlalchemy/issue/3034/…
zzzeek
66

Это работает в python 2 и 3 и немного чище, чем раньше, но требует SA> = 1.0.

from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType

# python2/3 compatible.
PY3 = str is not bytes
text = str if PY3 else unicode
int_type = int if PY3 else (int, long)
str_type = str if PY3 else (str, unicode)


class StringLiteral(String):
    """Teach SA how to literalize various things."""
    def literal_processor(self, dialect):
        super_processor = super(StringLiteral, self).literal_processor(dialect)

        def process(value):
            if isinstance(value, int_type):
                return text(value)
            if not isinstance(value, str_type):
                value = text(value)
            result = super_processor(value)
            if isinstance(result, bytes):
                result = result.decode(dialect.encoding)
            return result
        return process


class LiteralDialect(DefaultDialect):
    colspecs = {
        # prevent various encoding explosions
        String: StringLiteral,
        # teach SA about how to literalize a datetime
        DateTime: StringLiteral,
        # don't format py2 long integers to NULL
        NullType: StringLiteral,
    }


def literalquery(statement):
    """NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        statement = statement.statement
    return statement.compile(
        dialect=LiteralDialect(),
        compile_kwargs={'literal_binds': True},
    ).string

Демо-версия:

# coding: UTF-8
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        10 ** 20,  # a long integer
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print(literalquery(statement))


if __name__ == '__main__':
    test()

Дает этот вывод: (проверено в Python 2.7 и 3.4)

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
      '2015-06-24 18:09:29.042517', 3.14159, 100000000000000000000)
 LIMIT 1
bukzor
источник
2
Это круто ... Придется добавить это к некоторым библиотекам отладки, чтобы мы могли легко получить к ним доступ. Спасибо за проделанную работу. Я поражен, что это должно быть так сложно.
Кори О.
5
Я уверен, что это намеренно сложно, потому что новички испытывают искушение к cursor.execute () этой строки. Принцип согласия взрослых обычно используется в питоне.
Букзор
Очень полезно. Спасибо!
Клим
Очень мило на самом деле. Я взял на себя смелость и включил это в stackoverflow.com/a/42066590/2127439 , который охватывает SQLAlchemy v0.7.9 - v1.1.15, включая операторы INSERT и UPDATE (PY2 / PY3).
wolfmanx
очень хорошо. но это преобразование, как показано ниже. 1) запрос (Table) .filter (Table.Column1.is_ (False) к WHERE Column1 IS 0. 2) запрос (Table) .filter (Table.Column1.is_ (True) к WHERE Column1 IS 1. 3) запрос ( Таблица) .filter (Table.Column1 == func.any ([1,2,3])) в WHERE Column1 = любое ('[1,2,3]') выше преобразование неверно в синтаксисе.
Sekhar C
52

Учитывая, что то, что вы хотите, имеет смысл только при отладке, вы можете запустить SQLAlchemy echo=Trueдля регистрации всех запросов SQL. Например:

engine = create_engine(
    "mysql://scott:tiger@hostname/dbname",
    encoding="latin1",
    echo=True,
)

Это также может быть изменено только для одного запроса:

echo=False- если True, Engine регистрирует все операторы, а также repr()их списки параметров в регистраторе двигателей, который по умолчанию равен sys.stdout. echoАтрибут Engineможет быть изменен в любое время , чтобы включить ведение журнала и выключается. Если задана строка "debug", результирующие строки будут также выводиться на стандартный вывод. Этот флаг в конечном счете управляет регистратором Python; см. Настройка ведения журнала для получения информации о том, как напрямую настроить ведение журнала.

Источник: SQLAlchemy Engine Configuration

Если вы используете Flask, вы можете просто установить

app.config["SQLALCHEMY_ECHO"] = True

чтобы получить такое же поведение.

Ведран Шего
источник
6
Этот ответ заслуживает того, чтобы быть намного выше ... и для пользователей flask-sqlalchemyэтого должен быть принятый ответ.
JSO
25

Мы можем использовать метод компиляции для этой цели. Из документов :

from sqlalchemy.sql import text
from sqlalchemy.dialects import postgresql

stmt = text("SELECT * FROM users WHERE users.name BETWEEN :x AND :y")
stmt = stmt.bindparams(x="m", y="z")

print(stmt.compile(dialect=postgresql.dialect(),compile_kwargs={"literal_binds": True}))

Результат:

SELECT * FROM users WHERE users.name BETWEEN 'm' AND 'z'

Предупреждение из документов:

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

akshaynagpal
источник
13

Поэтому, опираясь на комментарии @zzzeek к коду @ bukzor, я придумал это, чтобы легко получить «симпатичный для печати» запрос:

def prettyprintable(statement, dialect=None, reindent=True):
    """Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement. The function can also receive a
    `sqlalchemy.orm.Query` object instead of statement.
    can 

    WARNING: Should only be used for debugging. Inlining parameters is not
             safe when handling user created data.
    """
    import sqlparse
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if dialect is None:
            dialect = statement.session.get_bind().dialect
        statement = statement.statement
    compiled = statement.compile(dialect=dialect,
                                 compile_kwargs={'literal_binds': True})
    return sqlparse.format(str(compiled), reindent=reindent)

Лично мне трудно читать код, который не имеет отступов, поэтому я использовал sqlparseдля переиндексации SQL. Может быть установлен с pip install sqlparse.

jmagnusson
источник
@bukzor Все значения работают, кроме datatime.now()одного, при использовании python 3 + sqlalchemy 1.0. Вы должны последовать совету @ zzzeek по созданию пользовательского TypeDecorator, чтобы он тоже работал.
Jmagnusson
Это слишком специфично. Дата и время не работают ни в одной комбинации Python и sqlalchemy. Также в py27 unicii unicii вызывает взрыв.
Букзор
Насколько я мог видеть, маршрут TypeDecorator требует от меня изменения определений таблиц, что не является разумным требованием для простого просмотра моих запросов. Я отредактировал свой ответ так, чтобы он был немного ближе к твоему и zzzeek, ​​но я выбрал путь собственного диалекта, который правильно ортогонален определениям таблиц.
Букзор
11

Этот код основан на блестящем существующем ответе @bukzor. Я просто добавил пользовательский рендер для datetime.datetimeтипа в Oracle TO_DATE().

Не стесняйтесь обновлять код в соответствии с вашей базой данных:

import decimal
import datetime

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if bind is None:
            bind = statement.session.get_bind(
                    statement._mapper_zero_or_none()
            )
        statement = statement.statement
    elif bind is None:
        bind = statement.bind 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )
        def render_literal_value(self, value, type_):
            """Render the value of a bind parameter as a quoted literal.

            This is used for statement sections that do not accept bind paramters
            on the target driver/database.

            This should be implemented by subclasses using the quoting services
            of the DBAPI.

            """
            if isinstance(value, basestring):
                value = value.replace("'", "''")
                return "'%s'" % value
            elif value is None:
                return "NULL"
            elif isinstance(value, (float, int, long)):
                return repr(value)
            elif isinstance(value, decimal.Decimal):
                return str(value)
            elif isinstance(value, datetime.datetime):
                return "TO_DATE('%s','YYYY-MM-DD HH24:MI:SS')" % value.strftime("%Y-%m-%d %H:%M:%S")

            else:
                raise NotImplementedError(
                            "Don't know how to literal-quote value %r" % value)            

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)
vvladymyrov
источник
22
Я не понимаю, почему люди из SA считают разумным, чтобы такая простая операция была такой сложной .
Букзор
Спасибо! render_literal_value работал хорошо для меня. Единственным изменением было то, что return "%s" % valueвместо return repr(value)секции float, int, long, потому что Python 22L22
выводил long,
Этот рецепт (как и оригинал) вызывает UnicodeDecodeError, если какое-либо строковое значение bindparam не может быть представлено в ascii. Я опубликовал суть, которая исправляет это.
gsakkis
1
"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S")в MySQL
Цитрак
1
@bukzor - я не помню, чтобы меня спрашивали, является ли вышеупомянутое «разумным», поэтому вы не можете утверждать, что я «верю», что это так - FWIW, это не так! :) пожалуйста, смотрите мой ответ.
zzzeek
8

Я хотел бы отметить, что приведенные выше решения не «просто работают» с нетривиальными запросами. Одна проблема, с которой я столкнулся, была более сложные типы, такие как pgsql ARRAYs, вызывающие проблемы. Я нашел решение, которое для меня, просто работает даже с массивами pgsql:

заимствовано из: https://gist.github.com/gsakkis/4572159

Похоже, что связанный код основан на более старой версии SQLAlchemy. Вы получите сообщение о том, что атрибут _mapper_zero_or_none не существует. Вот обновленная версия, которая будет работать с более новой версией, вы просто замените _mapper_zero_or_none на bind. Кроме того, здесь есть поддержка массивов pgsql:

# adapted from:
# https://gist.github.com/gsakkis/4572159
from datetime import date, timedelta
from datetime import datetime

from sqlalchemy.orm import Query


try:
    basestring
except NameError:
    basestring = str


def render_query(statement, dialect=None):
    """
    Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement.
    WARNING: This method of escaping is insecure, incomplete, and for debugging
    purposes only. Executing SQL statements with inline-rendered user values is
    extremely insecure.
    Based on http://stackoverflow.com/questions/5631078/sqlalchemy-print-the-actual-query
    """
    if isinstance(statement, Query):
        if dialect is None:
            dialect = statement.session.bind.dialect
        statement = statement.statement
    elif dialect is None:
        dialect = statement.bind.dialect

    class LiteralCompiler(dialect.statement_compiler):

        def visit_bindparam(self, bindparam, within_columns_clause=False,
                            literal_binds=False, **kwargs):
            return self.render_literal_value(bindparam.value, bindparam.type)

        def render_array_value(self, val, item_type):
            if isinstance(val, list):
                return "{%s}" % ",".join([self.render_array_value(x, item_type) for x in val])
            return self.render_literal_value(val, item_type)

        def render_literal_value(self, value, type_):
            if isinstance(value, long):
                return str(value)
            elif isinstance(value, (basestring, date, datetime, timedelta)):
                return "'%s'" % str(value).replace("'", "''")
            elif isinstance(value, list):
                return "'{%s}'" % (",".join([self.render_array_value(x, type_.item_type) for x in value]))
            return super(LiteralCompiler, self).render_literal_value(value, type_)

    return LiteralCompiler(dialect, statement).process(statement)

Проверено на двух уровнях вложенных массивов.

JamesHutchison
источник
Пожалуйста, покажите пример, как его использовать? Спасибо
slashdottir
from file import render_query; print(render_query(query))
Альфонсо Перес
Это единственный пример всей этой страницы, который работал на меня! Спасибо !
Fougerejo