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

Question

Mam obiekt kwerendy SQLAlchemy i chcę uzyskać tekst skompilowanej instrukcji SQL z wszystkimi powiązanymi parametrami (np. Brak %s lub innymi zmiennymi oczekującymi na związanie przez kompilator instrukcji lub silnik dialektowy MySQLdb, itp.).

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

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

próbowałam patrząc w query._params ale jest pusty dict. Napisałem własny kompilator, używając this example of the sqlalchemy.ext.compiler.compiles decorator, ale nawet tam, gdzie chcę dane, nadal istnieje %s.

Nie mogę się dowiedzieć, kiedy moje parametry zostaną pomieszane, aby utworzyć zapytanie; podczas badania obiektu zapytania są one zawsze pustym słownikiem (chociaż zapytanie wykonuje się poprawnie, a silnik drukuje po włączeniu rejestrowania echa).

Zaczynam otrzymywać komunikat, że SQLAlchemy nie chce, abym znał zapytanie bazowe, ponieważ łamie ogólną naturę interfejsu API ekspresji wszystkich różnych interfejsów API DB. Nie mam nic przeciwko, jeśli zapytanie zostanie wykonane, zanim się dowiem, co to było; Chcę po prostu wiedzieć!

Answer 1

This blog zapewnia zaktualizowaną odpowiedź.

Cytując z posta na blogu, jest to sugerowane i działa dla mnie.

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

gdzie q jest zdefiniowany jako:

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

lub po prostu jakiejkolwiek session.query().

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

Answer 2

ten powinien pracować 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) 

Answer 3

Chodzi o to, nigdy sqlalchemy łączy dane z zapytania. Zapytanie i dane są przekazywane osobno do sterownika bazy danych - interpolacja danych odbywa się w bazie danych.

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

Można zastosować podejście, w którym sam interpolujesz dane za pomocą zapytania (jak sugerował albertov poniżej), ale to nie jest to samo, co sqlalchemy.

Answer 4

Dla backendu MySQLdb zmodyfikowałem niesamowitą odpowiedź albertov (bardzo dziękuję!). Jestem pewien, że można je scalić, aby sprawdzić, czy comp.positional było prawdziwe, ale 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) 

Answer 5

myślę .statement byłoby ewentualnie rade: 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 

Answer 6

Można użyć zdarzenia z ConnectionEvents rodziny: after_cursor_execute lub before_cursor_execute.

W SQLAlchemy UsageRecipes przez @zzzeek można znaleźć ten 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żna uzyskać dostęp do wyciągu

Answer 7

documentation wykorzystuje literal_binds wydrukować zapytanie q tym parametry:

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

the above approach has the caveats that it is only supported for basic types, such as ints and strings, and furthermore if a bindparam() without a pre-set value is used directly, it won’t be able to stringify that either.

Answer 8

Następujące rozwiązanie wykorzystuje język wyrażeń SQLAlchemy i współpracuje z SQLAlchemy 1.1. To rozwiązanie nie miesza parametrów z zapytaniem (zgodnie z żądaniem oryginalnego autora), ale zapewnia sposób korzystania z modeli SQLAlchemy w celu generowania ciągów zapytań SQL i słowników parametrów dla różnych dialektów SQL. Przykład bazuje na tutorialu 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 produkować oświadczenie kwerendy przy użyciu wybierz funkcję .

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 zestawienie jest skompilowany przy użyciu podstawowego „o nazwie” wdrożenia, który jest kompatybilny z bazami danych SQL, takich jak SQLite i Oracle. Jeśli trzeba określić dialekt takiego jak PostgreSQL, można zrobić

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

Lub jeśli chcesz, aby wyraźnie określić dialekt jako SQLite, można zmienić paramstyle od „qmark” do „o nazwie”.

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

od obiektu zapytań, możemy wyodrębnić ciąg kwerendy i parametry zapytania

query_str = str(query) 
query_params = query.params 

i wreszcie wykonanie zapytania.

conn.execute(query_str, query_params) 

Answer 9

dla PostgreSQL backend przy użyciu psycopg2 można wykrywać zdarzenie do_execute, a następnie za pomocą kursora, oświadczenie oraz parametrów typu zmuszony wraz z Cursor.mogrify() do inline parametry. Możesz zwrócić wartość True, aby uniemożliwić rzeczywiste wykonanie 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 

wykorzystanie próbki:

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

Answer 10

pierwsze chciałbym poprzedzić stwierdzeniem, że zakładam, że robisz to głównie do celów debugowania - nie polecam próbuje zmodyfikować oświadczenie z zewnątrz Płynny interfejs API SQLAlchemy.

Niestety nie ma prostego sposobu na pokazanie skompilowanej instrukcji z uwzględnieniem parametrów zapytania. SQLAlchemy w rzeczywistości nie wstawia parametrów do instrukcji - są one passed into the database engine as a dictionary.Dzięki temu uchwyt biblioteki specyficznej dla bazy danych może być unikany znaków specjalnych i tym podobnych, aby uniknąć iniekcji SQL.

Ale można to zrobić w dwuetapowy sposób z łatwością. Aby uzyskać oświadczenie, można to zrobić jak już pokazano, i po prostu wydrukować zapytanie:

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

można dostać jeden krok bliżej query.statement, aby zobaczyć nazwy parametrów. (Uwaga :id_1 poniżej vs %s powyżej - naprawdę nie problem w tym bardzo prosty przykład, ale może być kluczem w bardziej skomplikowanych instrukcji.)

>>> print(query.statement) 
>>> print(query.statement.compile()) # reasonably 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żna uzyskać rzeczywiste wartości parametrów przez coraz własność params skompilowany stwierdzeniem:

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

To działało na backend MySQL przynajmniej; Spodziewam się, że jest on również wystarczająco ogólny dla PostgreSQL bez potrzeby używania psycopg2.