SQLAlchemy: wydrukuj rzeczywiste zapytanie


165

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?


1
Nie mam, ale prawdopodobnie możesz zbudować mniej delikatne rozwiązanie, korzystając z sqlalchemy.enginedziennika SQLAlchemy . Rejestruje zapytania i parametry wiązania, wystarczy zamienić symbole zastępcze powiązań na wartości z łatwo skonstruowanego ciągu zapytania SQL.
Simon,

@Simon: są dwa problemy z używaniem loggera: 1) drukuje tylko wtedy, gdy wykonywana jest instrukcja 2) Nadal musiałbym wykonać zamianę ciągu, z wyjątkiem tego przypadku, nie znałbym dokładnie ciągu szablonu powiązania , i musiałbym jakoś wyodrębnić go z tekstu zapytania, czyniąc rozwiązanie bardziej delikatnym.
bukzor

Nowy adres URL wygląda na docs.sqlalchemy.org/en/latest/faq/ ... do FAQ @ zzzeek.
Jim DeLaHunt

Odpowiedzi:


168

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)

2
To nie umieszcza cudzysłowów wokół łańcuchów i nie rozwiązuje niektórych związanych parametrów.
bukzor

1
druga połowa odpowiedzi została zaktualizowana o najnowsze informacje.
zzzeek 30.07.14

2
@zzzeek Dlaczego w sqlalchemy domyślnie nie są zawarte zapytania o ładnym druku? Lubię query.prettyprint(). Niezwykle łagodzi problem związany z debugowaniem przy dużych zapytaniach.
jmagnusson

2
@jmagnusson, ponieważ piękno jest w oku patrzącego :) Istnieje wiele haków (np. zdarzenie kursor_execute, filtry logowania w Pythonie @compilesitp.) dla dowolnej liczby pakietów firm trzecich, aby zaimplementować ładne systemy drukujące.
zzzeek

1
@buzkor re: limit, który został naprawiony w wersji 1.0 bitbucket.org/zzzeek/sqlalchemy/issue/3034/…
zzzeek

66

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

2
To jest niesamowite ... Będziemy musieli dodać to do niektórych bibliotek debugowania, abyśmy mogli łatwo uzyskać do nich dostęp. Dzięki za pracę nóg nad tym. Dziwię się, że musiało to być takie skomplikowane.
Corey O.

5
Jestem prawie pewien, że celowo jest to trudne, ponieważ nowicjusze są kuszeni do tego ciągu. Jednak zasada wyrażania zgody przez dorosłych jest powszechnie stosowana w Pythonie.
bukzor

Bardzo przydatne. Dzięki!
clime

Naprawdę bardzo ładnie. Pozwoliłem sobie na to i włączyłem to do stackoverflow.com/a/42066590/2127439 , który obejmuje SQLAlchemy v0.7.9 - v1.1.15, w tym instrukcje INSERT i UPDATE (PY2 / PY3).
wolfmanx

bardzo dobrze. ale czy konwertuje jak poniżej. 1) query (Table) .filter (Table.Column1.is_ (False) to WHERE Column1 IS 0. 2) query (Table) .filter (Table.Column1.is_ (True) to WHERE Column1 IS 1. 3) query ( Table) .filter (Table.Column1 == func.any ([1,2,3])) do WHERE Kolumna1 = dowolne ('[1,2,3]') powyższe konwersje są niepoprawne w składni.
Sekhar C

51

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.


6
Ta odpowiedź zasługuje na znacznie wyższą odpowiedź… i dla użytkowników flask-sqlalchemypowinna być odpowiedzią akceptowaną.
jso

25

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.


13

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.


@bukzor Wszystkie wartości działają z wyjątkiem datatime.now()jednej, gdy używasz Pythona 3 + sqlalchemy 1.0. Musiałbyś postępować zgodnie z radą @ zzzeek dotyczącą tworzenia niestandardowego TypeDecorator, aby ten również działał.
jmagnusson

To trochę zbyt szczegółowe. Data i godzina nie działa w żadnej kombinacji języka Python i sqlalchemy. Również w py27, nie-ascii unicode powoduje eksplozję.
bukzor

O ile mogłem zobaczyć, trasa TypeDecorator wymaga ode mnie zmiany definicji tabel, co nie jest rozsądnym wymogiem, aby po prostu zobaczyć moje zapytania. Zredagowałem swoją odpowiedź, aby była trochę bliższa twojej i zzzeek, ​​ale wybrałem ścieżkę niestandardowego dialektu, który jest właściwie ortogonalny do definicji tabel.
bukzor

11

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)

22
Nie rozumiem, dlaczego ludzie z SA uważają, że tak prosta operacja jest tak trudna , że rozsądne jest .
bukzor

Dziękuję Ci! render_literal_value działał dobrze dla mnie. Moją jedyną zmianą było: return "%s" % valuezamiast return repr(value)w sekcji typu float, int, long, ponieważ Python wyświetlał długie, a 22Lnie tylko22
OrganicPanda

Ten przepis (podobnie jak oryginał) wywołuje błąd UnicodeDecodeError, jeśli jakakolwiek wartość łańcucha bindparam nie jest reprezentowalna w ascii. Opublikowałem sedno, które to rozwiązuje.
gsakkis

1
"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S")w mysql
Zitrax

1
@bukzor - Nie przypominam sobie, żeby ktoś zapytał mnie, czy powyższe jest „rozsądne”, więc nie można powiedzieć, że „wierzę”, że tak jest - FWIW, tak nie jest! :) zobacz moją odpowiedź.
zzzeek

8

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.


Proszę pokazać przykład, jak go używać? Dziękuję
slashdottir

from file import render_query; print(render_query(query))
Alfonso Pérez

To jedyny przykład całej tej strony, który działał dla mnie! Dzięki !
fougerejo
Korzystając z naszej strony potwierdzasz, że przeczytałeś(-aś) i rozumiesz nasze zasady używania plików cookie i zasady ochrony prywatności.
Licensed under cc by-sa 3.0 with attribution required.