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

103

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ć!

cce
źródło

Odpowiedzi:

107

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

AndyBarr
źródło
2
Czy istnieje łatwy sposób uzyskania wartości w słowniku?
Damien
6
@Damien biorąc pod uwagę c = q.statement.compile(...), możesz po prostu dostaćc.params
Hannele
1
Post jest oznaczony tagiem mysql, więc szczegóły postgresql w tej odpowiedzi nie są tak naprawdę istotne.
Hannele,
4
Jeśli dobrze rozumiem OP, chce ostatecznego zapytania. Drukowanie z określeniem dialektu (tutaj postgres) nadal daje mi symbole zastępcze zamiast wartości dosłownych. @ Odpowiedź Matta załatwia sprawę. Uzyskanie kodu SQL z symbolami zastępczymi można łatwiej osiągnąć za pomocą as_scalar()-metody Query.
Patrick B.
1
@PatrickB. Zgadzam się. Odpowiedź Matta powinna być uważana za „prawidłową” odpowiedź. Osiągam ten sam wynik, po prostu robiąc str(q).
André C. Andersen
94

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.

Matt
źródło
Dziękuję Ci! To było niezwykle pomocne, pozwoliło mi bezboleśnie używać funkcji pandy read_sql!
Justin Palmer
24

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)
albertov
źródło
2
Dzięki za to! Niestety używam MySQL, więc mój dialekt jest „pozycyjny” i musi mieć raczej listę parametrów niż słownik. Obecnie próbuję zmusić Twój przykład do pracy z tym ..
cce
Proszę nie używać adaptw ten sposób. Przynajmniej za każdym razem wywołaj funkcję ready () na zwracanej z niej wartości, podając połączenie jako argument, aby mógł wykonać odpowiednie cytowanie.
Alex Gaynor,
@Alex: Jaki byłby poprawny sposób na prawidłowe cytowanie z psycopg? (oprócz wywoływania przygotować () na wartości zwracanej, które wydają się sugerować, nie jest optymalny)
Albertov
Przepraszam, myślę, że moje sformułowanie było złe, tak długo, jak wywołujesz obj.prepare (połączenie), powinno być dobrze. Dzieje się tak, ponieważ „dobre” interfejsy API, które libpq zapewnia do cytowania, wymagają połączenia (i zapewnia takie rzeczy, jak kodowanie dla łańcuchów znaków Unicode).
Alex Gaynor
1
Dzięki. Próbowałem powołanie preparena wartości zwracanej ale wydaje się, że nie ma tej metody: AttributeError: 'psycopg2._psycopg.AsIs' object has no attribute 'prepare'. Używam psycopg2 2.2.1 BTW
albertov
18

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)
cce
źródło
Niesamowite! Po prostu potrzebowałem, aby powiązana lista parametrów została wysłana do MySQL i zmodyfikowała powyższe, aby return tuple(params)działała jak urok! Zaoszczędziłeś mi niezliczonych godzin jazdy po wyjątkowo bolesnej drodze.
horcle_buzz
16

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.

nosklo
źródło
dlaczego to nie jest to samo? Rozumiem, że DB-API wykonuje transakcje, prawdopodobnie zmienia kolejność zapytań itp., Ale czy może bardziej modyfikować moje zapytanie?
cce
1
@cce: próbujesz znaleźć ostateczne zapytanie. SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC JEST ostatecznym pytaniem. Są %sone wysyłane do bazy danych przez sqlalchemy - sqlalchemy NIGDY nie umieszcza rzeczywistych danych w miejscu% s
nosklo
@cce: Niektóre moduły dbapi też tego nie robią - często robi to sama baza danych
nosklo
1
aha Rozumiem, co mówisz, dzięki - zagłębiając się dalej sqlalchemy.dialects.mysql.mysqldb, do_executemany()przekazuje instrukcję i parametry osobno do kursora MySQLdb. yay indirection!
cce
11

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.

Hannele
źródło
Z debuggera PyCharm, działały dla mnie następujące funkcje
Ben
Ciekawe, że SQLAlchemy może się nieco zmienić od czasu napisania tej odpowiedzi.
Hannele
10

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"}}'}
rektalogiczny
źródło
4

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 )
eric
źródło
W jaki sposób ta odpowiedź jest lepsza / inna niż ta opublikowana przez AndyBarra 2 lata wcześniej?
Piotr Dobrogost
Odpowiedź AndyBarra zawiera przykład generowania instrukcji zapytania z DBSession, podczas gdy ta odpowiedź zawiera przykład wykorzystujący deklaratywne API i metodę select. Jeśli chodzi o zestawianie zapytania w określonym dialekcie, odpowiedzi są takie same. Używam SQLAlchemy do generowania surowych zapytań, a następnie wykonuję je za pomocą adbapi Twistera. W tym przypadku przydatna jest wiedza, jak skompilować zapytanie bez sesji oraz wyodrębnić ciąg zapytania i parametry.
eric
3

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

Alex Bender
źródło
2

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())
David K. Hess
źródło
0

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
user2757902
źródło
Instrukcja nie pokazuje, jakie są parametry, jeśli masz skonfigurowane jakieś rodzaje filtrów.
Hannele,