Jak uzyskać surowe, skompilowane zapytanie SQL z wyrażenia SQLAlchemy?

Mam obiekt zapytania SQLAlchemy i chcę uzyskać tekst skompilowanej instrukcji SQL ze wszystkimi powiązanymi parametrami (np. Brak %slub inne zmienne oczekujące na powiązanie przez kompilator instrukcji lub silnik dialektu MySQLdb itp.).

Wywołanie str()zapytania ujawnia coś takiego:

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

Próbowałem szukać w query._params, ale jest to pusty dykt. Napisałem swój własny kompilator, korzystając z tego przykładu sqlalchemy.ext.compiler.compilesdekoratora, ale nawet stwierdzenie tam nadal zawiera %sdane.

Nie mogę się do końca dowiedzieć, kiedy moje parametry są mieszane, aby utworzyć zapytanie; podczas badania obiektu zapytania są one zawsze pustym słownikiem (chociaż zapytanie wykonuje się dobrze, a silnik wypisuje je po włączeniu logowania echa).

Zaczynam otrzymywać komunikat, że SQLAlchemy nie chce, żebym znało zapytanie, ponieważ łamie on ogólny charakter interfejsu API wyrażeń we wszystkich różnych DB-API. Nie mam nic przeciwko, jeśli zapytanie zostanie wykonane, zanim dowiem się, co to było; Chcę po prostu wiedzieć!

Ten blog zawiera zaktualizowaną odpowiedź.

Cytując z posta na blogu, jest to zasugerowane i zadziałało dla mnie.

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

Gdzie q definiuje się jako:

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

Lub po prostu dowolny rodzaj session.query ().

Dzięki Nicolas Cadou za odpowiedź! Mam nadzieję, że pomoże to innym, którzy tu szukają.

Dokumentacja używa literal_bindsdo wydrukowania zapytanie qłącznie z parametrami:

print(q.statement.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 bindparam () bez wstępnie ustawionej wartości jest używana bezpośrednio, nie będzie w stanie tego również określić.

Dokumentacja zawiera również to ostrzeżenie:

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 programowego wywoływania instrukcji SQL innych niż DDL względem relacyjnej bazy danych.

To powinno działać z 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)

Dla zaplecza MySQLdb zmodyfikowałem trochę niesamowitą odpowiedź Alberta (dzięki wielkie!). Jestem pewien, że można je scalić, aby sprawdzić, czy comp.positionalbyło, Trueale to nieco wykracza poza zakres tego pytania.

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)

Rzecz w tym, że sqlalchemy nigdy nie miesza danych z zapytaniem. Zapytanie i dane są przekazywane oddzielnie do podstawowego sterownika bazy danych - interpolacja danych odbywa się w bazie danych.

Sqlalchemy przekazuje zapytanie str(myquery)do bazy danych, tak jak widzieliśmy , a wartości zostaną umieszczone w osobnej krotce.

Możesz użyć jakiegoś podejścia, w którym sam interpolujesz dane za pomocą zapytania (jak zasugerował albertov poniżej), ale to nie to samo, co wykonuje sqlalchemy.

Najpierw pozwolę sobie powiedzieć, że zakładam, że robisz to głównie w celu debugowania - nie polecałbym próbować modyfikować instrukcji poza API płynnym SQLAlchemy.

Niestety wydaje się, że nie ma prostego sposobu na pokazanie skompilowanej instrukcji z dołączonymi parametrami zapytania. SQLAlchemy w rzeczywistości nie umieszcza parametrów w instrukcji - są one przekazywane do silnika bazy danych jako słownik . Dzięki temu biblioteka specyficzna dla bazy danych może obsługiwać takie rzeczy, jak unikanie znaków specjalnych w celu uniknięcia wstrzyknięcia SQL.

Ale możesz to zrobić w dwuetapowym procesie dość łatwo. Aby uzyskać wyciąg, możesz zrobić to, co już pokazałeś i po prostu wydrukować zapytanie:

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

Możesz podejść o krok bliżej dzięki query.statement, aby zobaczyć nazwy parametrów. Uwaga :id_1poniżej i %spowyżej - nie jest to problem w tym bardzo prostym przykładzie, ale może być kluczowy w bardziej skomplikowanym stwierdzeniu.

>>> 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;

Następnie możesz uzyskać rzeczywiste wartości parametrów, pobierając paramswłaściwość skompilowanej instrukcji:

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

To działało przynajmniej dla zaplecza MySQL; Spodziewałbym się, że jest również wystarczająco ogólny dla PostgreSQL bez konieczności używania psycopg2.

Dla zaplecza postgresql używającego psycopg2, możesz nasłuchiwać do_executezdarzenia, a następnie użyć kursora, instrukcji i wpisać wymuszone parametry wraz z Cursor.mogrify()wbudowanymi parametrami. Możesz zwrócić True, aby zapobiec faktycznemu wykonaniu zapytania.

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

Przykładowe użycie:

>>> 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"}}'}

Poniższe rozwiązanie używa języka SQLAlchemy Expression Language i współpracuje z SQLAlchemy 1.1. To rozwiązanie nie miesza parametrów z zapytaniem (zgodnie z żądaniem pierwotnego autora), ale zapewnia sposób używania modeli SQLAlchemy do generowania ciągów zapytań SQL i słowników parametrów dla różnych dialektów SQL. Przykład jest oparty na samouczku http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Biorąc pod uwagę klasę,

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

możemy utworzyć instrukcję zapytania za pomocą funkcji select .

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

Następnie możemy skompilować instrukcję do obiektu zapytania.

query = statement.compile()

Domyślnie instrukcja jest kompilowana przy użyciu podstawowej „nazwanej” implementacji, która jest zgodna z bazami danych SQL, takimi jak SQLite i Oracle. Jeśli potrzebujesz określić dialekt, taki jak PostgreSQL, możesz to zrobić

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

Lub jeśli chcesz jawnie określić dialekt jako SQLite, możesz zmienić styl parametru z „qmark” na „named”.

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

Z obiektu zapytania możemy wyodrębnić ciąg zapytania i parametry zapytania

query_str = str(query)
query_params = query.params

i na koniec wykonaj zapytanie.

conn.execute( query_str, query_params )

Możesz użyć zdarzeń z rodziny ConnectionEvents : after_cursor_executelub before_cursor_execute.

W sqlalchemy UsageRecipes autorstwa @zzzeek można znaleźć następujący przykład:

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

Tutaj możesz uzyskać dostęp do swojego wyciągu

Tak więc, łącząc wiele małych fragmentów tych różnych odpowiedzi, wymyśliłem to, czego potrzebowałem: prosty zestaw kodu do wrzucenia i czasami, ale niezawodnie (tj. Obsługuje wszystkie typy danych), pobieram dokładny, skompilowany kod SQL wysłany do mojego Backend Postgres, po prostu odpytując samo zapytanie:

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

Myślę, że .statement prawdopodobnie załatwi sprawę: 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