Wie erhalte ich eine rohe, kompilierte SQL-Abfrage aus einem SQLAlchemy-Ausdruck?

Ich habe ein SQLAlchemy-Abfrageobjekt und möchte den Text der kompilierten SQL-Anweisung mit allen gebundenen Parametern %sabrufen (z. B. keine oder andere Variablen, die darauf warten, vom Anweisungscompiler oder der MySQLdb-Dialekt-Engine usw. gebunden zu werden).

Das Aufrufen str()der Abfrage zeigt ungefähr Folgendes:

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

Ich habe versucht, in query._params zu suchen, aber es ist ein leeres Diktat. Ich habe meinen eigenen Compiler anhand dieses Beispiels des sqlalchemy.ext.compiler.compilesDekorateurs geschrieben, aber selbst die dortige Anweisung enthält noch %sDaten.

Ich kann nicht genau herausfinden, wann meine Parameter zum Erstellen der Abfrage gemischt werden. Bei der Untersuchung des Abfrageobjekts handelt es sich immer um ein leeres Wörterbuch (obwohl die Abfrage einwandfrei ausgeführt wird und die Engine sie ausdruckt, wenn Sie die Echoanmeldung aktivieren).

Ich bekomme allmählich die Meldung, dass SQLAlchemy nicht möchte, dass ich die zugrunde liegende Abfrage kenne, da dies die allgemeine Natur der Schnittstelle der Ausdrucks-API in allen verschiedenen DB-APIs beeinträchtigt. Es macht mir nichts aus, wenn die Abfrage ausgeführt wird, bevor ich herausgefunden habe, was es war. Ich will nur wissen!

Dieser Blog bietet eine aktualisierte Antwort.

Zitat aus dem Blog-Beitrag, dies wird vorgeschlagen und hat für mich funktioniert.

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

Wobei q definiert ist als:

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

Oder einfach jede Art von session.query ().

Vielen Dank an Nicolas Cadou für die Antwort! Ich hoffe, es hilft anderen, die hier suchen.

In der Dokumentation wird literal_bindseine Abfrage qmit folgenden Parametern gedruckt :

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

Der obige Ansatz hat die Einschränkungen, dass er nur für Basistypen wie Ints und Strings unterstützt wird. Wenn ein bindparam () ohne voreingestellten Wert direkt verwendet wird, kann er dies auch nicht stringifizieren.

In der Dokumentation wird auch diese Warnung ausgegeben:

Verwenden Sie diese Technik niemals mit Zeichenfolgeninhalten, die von nicht vertrauenswürdigen Eingaben empfangen werden, z. B. von Webformularen oder anderen Benutzereingabeanwendungen. Die Funktionen von SQLAlchemy zum Erzwingen von Python-Werten in direkte SQL-Zeichenfolgenwerte sind nicht gegen nicht vertrauenswürdige Eingaben geschützt und validieren den übergebenen Datentyp nicht. Verwenden Sie immer gebundene Parameter, wenn Sie programmgesteuert Nicht-DDL-SQL-Anweisungen für eine relationale Datenbank aufrufen.

Dies sollte mit Sqlalchemy> = 0,6 funktionieren

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)

Für das MySQLdb-Backend habe ich Albertovs großartige Antwort (vielen Dank!) Ein wenig geändert. Ich bin sicher , sie könnten zusammengeführt werden zu überprüfen , ob comp.positionalwar , Trueaber das ist etwas über den Rahmen dieser Frage.

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)

Die Sache ist, dass sqlalchemy die Daten niemals mit Ihrer Abfrage mischt. Die Abfrage und die Daten werden separat an Ihren zugrunde liegenden Datenbanktreiber übergeben - die Interpolation der Daten erfolgt in Ihrer Datenbank.

Sqlalchemy übergibt die Abfrage, wie Sie sie gesehen haben, str(myquery)an die Datenbank, und die Werte werden in einem separaten Tupel gespeichert.

Sie könnten einen Ansatz verwenden, bei dem Sie die Daten mit der Abfrage selbst interpolieren (wie unten von albertov vorgeschlagen), aber das ist nicht dasselbe, was sqlalchemy ausführt.

Für das Postgresql-Backend mit psycopg2 können Sie auf das do_executeEreignis warten und dann den Cursor, die Anweisung und den Typ erzwungene Parameter zusammen mit verwenden, Cursor.mogrify()um die Parameter zu inline. Sie können True zurückgeben, um die tatsächliche Ausführung der Abfrage zu verhindern.

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

Beispielnutzung:

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

Lassen Sie mich zunächst sagen, dass ich davon ausgehe, dass Sie dies hauptsächlich zu Debugging-Zwecken tun. Ich würde nicht empfehlen, die Anweisung außerhalb der fließenden SQLAlchemy-API zu ändern.

Leider scheint es keine einfache Möglichkeit zu geben, die kompilierte Anweisung mit den enthaltenen Abfrageparametern anzuzeigen. SQLAlchemy fügt die Parameter nicht in die Anweisung ein - sie werden als Wörterbuch an das Datenbankmodul übergeben . Auf diese Weise kann die datenbankspezifische Bibliothek beispielsweise das Escapezeichen von Sonderzeichen verarbeiten, um eine SQL-Injection zu vermeiden.

Sie können dies jedoch relativ einfach in zwei Schritten tun. Um die Anweisung zu erhalten, können Sie das tun, was Sie bereits gezeigt haben, und einfach die Abfrage ausdrucken:

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

Mit query.statement können Sie einen Schritt näher kommen, um die Parameternamen anzuzeigen. (Anmerkung :id_1unten vs %soben - in diesem sehr einfachen Beispiel kein wirkliches Problem, könnte aber der Schlüssel zu einer komplizierteren Aussage sein.)

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

Anschließend können Sie die tatsächlichen Werte der Parameter abrufen, indem Sie die paramsEigenschaft der kompilierten Anweisung abrufen:

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

Dies funktionierte zumindest für ein MySQL-Backend. Ich würde erwarten, dass es auch allgemein genug für PostgreSQL ist, ohne es verwenden zu müssen psycopg2.

Die folgende Lösung verwendet die SQLAlchemy Expression Language und funktioniert mit SQLAlchemy 1.1. Diese Lösung mischt die Parameter nicht mit der Abfrage (wie vom ursprünglichen Autor angefordert), bietet jedoch eine Möglichkeit, mithilfe von SQLAlchemy-Modellen SQL-Abfragezeichenfolgen und Parameterwörterbücher für verschiedene SQL-Dialekte zu generieren. Das Beispiel basiert auf dem Tutorial http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Angesichts der Klasse,

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

Mit der Funktion select können wir eine Abfrageanweisung erstellen .

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

Als nächstes können wir die Anweisung in ein Abfrageobjekt kompilieren.

query = statement.compile()

Standardmäßig wird die Anweisung mit einer grundlegenden 'Named'-Implementierung kompiliert, die mit SQL-Datenbanken wie SQLite und Oracle kompatibel ist. Wenn Sie einen Dialekt wie PostgreSQL angeben müssen, können Sie dies tun

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

Wenn Sie den Dialekt explizit als SQLite angeben möchten, können Sie den Parameterstil von "qmark" in "named" ändern.

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

Aus dem Abfrageobjekt können wir die Abfragezeichenfolge und die Abfrageparameter extrahieren

query_str = str(query)
query_params = query.params

und führen Sie schließlich die Abfrage aus.

conn.execute( query_str, query_params )

Sie können Ereignisse aus der ConnectionEvents- Familie verwenden: after_cursor_executeoder before_cursor_execute.

In sqlalchemy UsageRecipes von @zzzeek finden Sie dieses Beispiel:

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

Hier können Sie auf Ihre Abrechnung zugreifen

Als ich viele kleine Teile dieser verschiedenen Antworten zusammenstellte, kam ich auf das, was ich brauchte: einen einfachen Satz Code, den ich einlegen und gelegentlich, aber zuverlässig (dh alle Datentypen handhaben), um die genaue, kompilierte SQL zu erhalten, die an meine gesendet wurde Postgres-Backend durch einfaches Abfragen der Abfrage selbst:

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

Ich denke, .statement würde möglicherweise den Trick tun: 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