SQLAlchemy: Druckt die eigentliche Abfrage

Ich möchte wirklich in der Lage sein, gültiges SQL für meine Anwendung, einschließlich Werte, auszudrucken, anstatt Parameter zu binden, aber es ist nicht offensichtlich, wie dies in SQLAlchemy gemacht wird (ich bin mir ziemlich sicher, dass dies beabsichtigt ist).

Hat jemand dieses Problem allgemein gelöst?

In den allermeisten Fällen ist die "Stringifizierung" einer SQLAlchemy-Anweisung oder -Abfrage so einfach wie:

print str(statement)

Dies gilt sowohl für ein ORM Queryals auch für eine select()oder andere Aussage.

Hinweis : Die folgende detaillierte Antwort wird in der sqlalchemy-Dokumentation beibehalten .

Um die Anweisung so zu kompilieren, dass sie für einen bestimmten Dialekt oder eine bestimmte Engine kompiliert wurde, können Sie diese an compile () übergeben , wenn die Anweisung selbst noch nicht an eine gebunden ist :

print statement.compile(someengine)

oder ohne Motor:

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

Wenn wir ein ORM- QueryObjekt erhalten, müssen compile()wir zuerst nur auf den .statement- Accessor zugreifen , um an die Methode zu gelangen :

statement = query.statement
print statement.compile(someengine)

In Bezug auf die ursprüngliche Bestimmung, dass gebundene Parameter in die endgültige Zeichenfolge "eingebunden" werden sollen, besteht die Herausforderung darin, dass SQLAlchemy normalerweise nicht damit beauftragt ist, da dies vom Python-DBAPI angemessen behandelt wird, ganz zu schweigen von der Umgehung gebundener Parameter wahrscheinlich die am häufigsten genutzten Sicherheitslücken in modernen Webanwendungen. SQLAlchemy ist unter bestimmten Umständen, z. B. bei der Emission von DDL, nur eingeschränkt in der Lage, diese Stringifizierung durchzuführen. Um auf diese Funktionalität zuzugreifen, kann das Flag 'literal_binds' verwendet werden, das an Folgendes übergeben wird 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})

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

Um das Inline-Literal-Rendering für nicht unterstützte Typen zu unterstützen, implementieren Sie ein TypeDecoratorfür den Zieltyp, das eine TypeDecorator.process_literal_paramMethode enthält:

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

Produktion produzieren wie:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)

Dies funktioniert in Python 2 und 3 und ist etwas sauberer als zuvor, erfordert jedoch 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

Demo:

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

Gibt diese Ausgabe: (getestet in Python 2.7 und 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

Da das, was Sie möchten, nur beim Debuggen sinnvoll ist, können Sie SQLAlchemy mit starten echo=True, um alle SQL-Abfragen zu protokollieren. Beispielsweise:

engine = create_engine(
    "mysql://scott:tiger@hostname/dbname",
    encoding="latin1",
    echo=True,
)

Dies kann auch nur für eine einzelne Anfrage geändert werden:

echo=False- Wenn Trueja, protokolliert die Engine alle Anweisungen sowie eine repr()ihrer Parameterlisten im Engineers-Logger, der standardmäßig verwendet wird sys.stdout. Das echoAttribut von Enginekann jederzeit geändert werden, um die Protokollierung ein- und auszuschalten. Wenn die Zeichenfolge festgelegt ist "debug", werden die Ergebniszeilen auch in der Standardausgabe gedruckt. Dieses Flag steuert letztendlich einen Python-Logger. siehe Konfigurieren der Protokollierung für Informationen zum Konfigurieren Protokollierung direkt.

Quelle: SQLAlchemy Engine-Konfiguration

Bei Verwendung mit Flask können Sie einfach einstellen

app.config["SQLALCHEMY_ECHO"] = True

das gleiche Verhalten zu bekommen.

Zu diesem Zweck können wir die Kompilierungsmethode verwenden. Aus den Dokumenten :

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

Ergebnis:

SELECT * FROM users WHERE users.name BETWEEN 'm' AND 'z'

Warnung aus Dokumenten:

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.

Aufbauend auf den Kommentaren von @ zzzeek zum Code von @ bukzor habe ich mir diese ausgedacht, um leicht eine "ziemlich druckbare" Abfrage zu erhalten:

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)

Ich persönlich habe sqlparseSchwierigkeiten, Code zu lesen, der nicht eingerückt ist, daher habe ich SQL neu eingerückt. Es kann mit installiert werden pip install sqlparse.

Dieser Code basiert auf einer brillanten Antwort von @bukzor. Ich habe gerade ein benutzerdefiniertes Rendering für den datetime.datetimeTyp in Oracle hinzugefügt TO_DATE().

Fühlen Sie sich frei, den Code entsprechend Ihrer Datenbank zu aktualisieren:

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)

Ich möchte darauf hinweisen, dass die oben angegebenen Lösungen nicht "nur" mit nicht trivialen Abfragen funktionieren. Ein Problem, auf das ich gestoßen bin, waren kompliziertere Typen, wie z. B. pgsql ARRAYs, die Probleme verursachen. Ich habe eine Lösung gefunden, die für mich sogar mit pgsql ARRAYs funktioniert hat:

ausgeliehen von: https://gist.github.com/gsakkis/4572159

Der verknüpfte Code scheint auf einer älteren Version von SQLAlchemy zu basieren. Sie erhalten eine Fehlermeldung, dass das Attribut _mapper_zero_or_none nicht vorhanden ist. Hier ist eine aktualisierte Version, die mit einer neueren Version funktioniert. Sie ersetzen einfach _mapper_zero_or_none durch bind. Darüber hinaus werden pgsql-Arrays unterstützt:

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

Getestet auf zwei Ebenen verschachtelter Arrays.