SQLAlchemy: wydrukuj rzeczywiste zapytanie

Naprawdę chciałbym móc wydrukować prawidłowy kod SQL dla mojej aplikacji, w tym wartości, zamiast powiązać parametry, ale nie jest oczywiste, jak to zrobić w SQLAlchemy (z założenia, jestem prawie pewien).

Czy ktoś rozwiązał ten problem w sposób ogólny?

W większości przypadków „stringification” instrukcji lub zapytania SQLAlchemy jest tak proste, jak:

print str(statement)

Dotyczy to zarówno ORM, Queryjak i każdego select()innego oświadczenia.

Uwaga : następująca szczegółowa odpowiedź jest przechowywana w dokumentacji sqlalchemy .

Aby otrzymać instrukcję skompilowaną do określonego dialektu lub silnika, jeśli sama instrukcja nie jest już powiązana z żadnym, możesz przekazać ją do metody compile () :

print statement.compile(someengine)

lub bez silnika:

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

Gdy otrzymamy Queryobiekt ORM , aby uzyskać dostęp do compile()metody, musimy najpierw uzyskać dostęp do metody dostępu .statement :

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

jeśli chodzi o pierwotne zastrzeżenie, że powiązane parametry mają być „wstawiane” do końcowego ciągu, wyzwaniem jest tutaj to, że SQLAlchemy normalnie nie ma tego zadania, ponieważ jest to odpowiednio obsługiwane przez Python DBAPI, nie wspominając o pomijaniu powiązanych parametrów jest prawdopodobnie najczęściej wykorzystywane luki bezpieczeństwa w nowoczesnych aplikacjach internetowych. SQLAlchemy ma ograniczone możliwości wykonywania tego ciągu w pewnych okolicznościach, takich jak emisja DDL. Aby uzyskać dostęp do tej funkcjonalności, można użyć flagi „literal_binds” przekazanej do 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})

powyższe podejście ma zastrzeżenia, że ​​jest obsługiwane tylko dla podstawowych typów, takich jak ints i stringi, a ponadto, jeśli a bindparam bez wstępnie ustawionej wartości jest używane bezpośrednio, nie będzie w stanie tego również określić.

Aby obsługiwać renderowanie literału wbudowanego dla typów nieobsługiwanych, zaimplementuj a TypeDecoratordla typu docelowego, który zawiera TypeDecorator.process_literal_parammetodę:

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})
)

wytwarzanie wyników takich jak:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)

Działa to w Pythonie 2 i 3 i jest nieco czystsze niż wcześniej, ale wymaga 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

Próbny:

# 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()

Daje to wyjście: (testowane w Pythonie 2.7 i 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

Biorąc pod uwagę, że to, co chcesz, ma sens tylko podczas debugowania, możesz uruchomić SQLAlchemy z echo=True, aby rejestrować wszystkie zapytania SQL. Na przykład:

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

Można to również zmodyfikować dla jednego żądania:

echo=False- jeśli Truesilnik będzie logował wszystkie instrukcje, a także listę repr()ich parametrów do rejestratora silników, który domyślnie sys.stdout. echoAtrybutem Enginemogą być zmienione w każdej chwili, aby włączyć rejestrowanie i wyłączać. Jeśli ustawione na łańcuch "debug", wiersze wyników będą również drukowane na standardowe wyjście. Ta flaga ostatecznie kontroluje program rejestrujący Pythona; zobacz Konfigurowanie rejestrowania, aby uzyskać informacje na temat bezpośredniego konfigurowania rejestrowania.

Źródło: Konfiguracja silnika SQLAlchemy

Jeśli jest używany z Flaskiem, możesz po prostu ustawić

app.config["SQLALCHEMY_ECHO"] = True

uzyskać takie samo zachowanie.

W tym celu możemy użyć metody kompilacji . Z dokumentów :

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}))

Wynik:

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

Ostrzeżenie od dokumentów:

Nigdy nie używaj tej techniki w przypadku zawartości ciągu znaków otrzymanej z niezaufanych danych wejściowych, takich jak formularze internetowe lub inne aplikacje wprowadzane przez użytkownika. Funkcje SQLAlchemy do przekształcania wartości Pythona w bezpośrednie wartości ciągów SQL nie są zabezpieczone przed niezaufanymi danymi wejściowymi i nie sprawdzają poprawności typu przekazywanych danych. Zawsze używaj parametrów powiązanych podczas programistycznego wywoływania instrukcji SQL innych niż DDL względem relacyjnej bazy danych.

Opierając się na komentarzach @ zzzeek do kodu @ bukzor, wymyśliłem to, aby łatwo uzyskać zapytanie „ładne do wydrukowania”:

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)

Osobiście mam trudności z odczytaniem kodu, który nie jest wcięty, więc użyłem sqlparseponownie kodu SQL. Można go zainstalować z pip install sqlparse.

Ten kod jest oparty na genialnej istniejącej odpowiedzi od @bukzor. Właśnie dodałem niestandardowy render dla datetime.datetimetypu do Oracle's TO_DATE().

Zapraszam do aktualizacji kodu, aby pasował do Twojej bazy danych:

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)

Zwracam uwagę, że powyższe rozwiązania nie „działają” tylko przy nietrywialnych zapytaniach. Jednym z problemów, na które się natknąłem, były bardziej skomplikowane typy, takie jak pgsql ARRAY powodujące problemy. Znalazłem rozwiązanie, które dla mnie działało nawet z pgsql ARRAY:

zapożyczone z: https://gist.github.com/gsakkis/4572159

Wydaje się, że powiązany kod jest oparty na starszej wersji SQLAlchemy. Pojawi się błąd informujący, że atrybut _mapper_zero_or_none nie istnieje. Oto zaktualizowana wersja, która będzie działać z nowszą wersją, po prostu zamień _mapper_zero_or_none na bind. Dodatkowo obsługuje to tablice 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)

Przetestowano na dwóch poziomach zagnieżdżonych tablic.