SQLAlchemy: affiche la requête réelle

165

J'aimerais vraiment pouvoir imprimer du SQL valide pour mon application, y compris des valeurs, plutôt que des paramètres de liaison, mais ce n'est pas évident de le faire dans SQLAlchemy (de par sa conception, j'en suis assez sûr).

Quelqu'un at-il résolu ce problème de manière générale?

Bukzor
la source
1
Je ne l'ai pas fait, mais vous pourriez probablement créer une solution moins fragile en exploitant le sqlalchemy.enginejournal de SQLAlchemy . Il enregistre les requêtes et les paramètres de liaison, il vous suffit de remplacer les espaces réservés de liaison par les valeurs d'une chaîne de requête SQL facilement construite.
Simon
@Simon: il y a deux problèmes avec l'utilisation de l'enregistreur: 1) il ne s'imprime que lorsqu'une instruction est en cours d' exécution 2) je devrais encore faire un remplacement de chaîne, sauf dans ce cas, je ne connais pas exactement la chaîne du modèle de liaison , et je devrais en quelque sorte l'analyser hors du texte de la requête, ce qui rendrait la solution plus fragile.
bukzor
La nouvelle URL semble être docs.sqlalchemy.org/en/latest/faq/… pour la FAQ de @ zzzeek.
Jim DeLaHunt

Réponses:

168

Dans la grande majorité des cas, la "stringification" d'une instruction ou d'une requête SQLAlchemy est aussi simple que:

print str(statement)

Cela s'applique à la fois à un ORM Queryet à toute select()instruction ou autre.

Remarque : la réponse détaillée suivante est maintenue dans la documentation de sqlalchemy .

Pour obtenir l'instruction compilée dans un dialecte ou un moteur spécifique, si l'instruction elle-même n'est pas déjà liée à un, vous pouvez la transmettre à compile () :

print statement.compile(someengine)

ou sans moteur:

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

Lorsqu'on lui donne un Queryobjet ORM , pour accéder à la compile()méthode, il suffit d'accéder d' abord à l' accesseur .statement :

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

en ce qui concerne la stipulation d'origine selon laquelle les paramètres liés doivent être "intégrés" dans la chaîne finale, le défi ici est que SQLAlchemy n'est normalement pas chargé de cela, car cela est géré de manière appropriée par le Python DBAPI, sans parler du contournement des paramètres liés. probablement les failles de sécurité les plus largement exploitées dans les applications Web modernes. SQLAlchemy a une capacité limitée à effectuer cette stringification dans certaines circonstances telles que l'émission de DDL. Pour accéder à cette fonctionnalité, on peut utiliser le drapeau 'literal_binds', passé à 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})

l'approche ci-dessus a la mise en garde qu'elle n'est prise en charge que pour les types de base, tels que les ints et les chaînes, et de plus si un bindparam sans valeur prédéfinie est utilisé directement, il ne pourra pas non plus le stringify.

Pour prendre en charge le rendu littéral en ligne pour les types non pris en charge, implémentez un TypeDecoratorpour le type cible qui inclut une TypeDecorator.process_literal_paramméthode:

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

produisant une sortie comme:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)
zzzeek
la source
2
Cela ne met pas de guillemets autour des chaînes et ne résout pas certains paramètres liés.
bukzor le
1
la seconde moitié de la réponse a été mise à jour avec les dernières informations.
zzzeek
2
@zzzeek Pourquoi les requêtes jolies imprimantes ne sont-elles pas incluses par défaut dans sqlalchemy? Comme query.prettyprint(). Cela facilite considérablement le débogage avec les grosses requêtes.
jmagnusson
2
@jmagnusson parce que la beauté est dans l'œil du spectateur :) Il y a de nombreux hooks (par exemple, un événement cursor_execute, des filtres de journalisation Python @compiles, etc.) pour n'importe quel nombre de packages tiers pour implémenter de jolis systèmes d'impression.
zzzeek
1
@buzkor re: limite qui a été corrigée dans la version 1.0 bitbucket.org/zzzeek/sqlalchemy/issue/3034/…
zzzeek
66

Cela fonctionne en python 2 et 3 et est un peu plus propre qu'avant, mais nécessite 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

Démo:

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

Donne cette sortie: (testé en python 2.7 et 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
Bukzor
la source
2
C'est génial ... Devra ajouter ceci à certaines bibliothèques de débogage pour que nous puissions y accéder facilement. Merci d'avoir fait le jeu de jambes sur celui-ci. Je suis étonné que ce soit si compliqué.
Corey O.24
5
Je suis presque sûr que c'est intentionnellement difficile, car les débutants sont tentés de curseur.execute () cette chaîne. Le principe d'adultes consentants est cependant couramment utilisé en python.
bukzor
Très utile. Merci!
clime
Très beau effectivement. J'ai pris la liberté et incorporé ceci dans stackoverflow.com/a/42066590/2127439 , qui couvre SQLAlchemy v0.7.9 - v1.1.15, y compris les instructions INSERT et UPDATE (PY2 / PY3).
wolfmanx
très agréable. mais est-ce la conversion comme ci-dessous. 1) query (Table) .filter (Table.Column1.is_ (False) to WHERE Column1 IS 0. 2) query (Table) .filter (Table.Column1.is_ (True) to WHERE Column1 IS 1. 3) query ( Table) .filter (Table.Column1 == func.any ([1,2,3])) à WHERE Column1 = any ('[1,2,3]') ci-dessus les conversions sont incorrectes dans la syntaxe.
Sekhar C
52

Étant donné que ce que vous voulez n'a de sens que lors du débogage, vous pouvez démarrer SQLAlchemy avec echo=True, pour enregistrer toutes les requêtes SQL. Par exemple:

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

Cela peut également être modifié pour une seule demande:

echo=False- si True, le moteur enregistre toutes les instructions ainsi qu'une repr()de leurs listes de paramètres dans le journal des moteurs, qui est par défaut sys.stdout. L' echoattribut de Enginepeut être modifié à tout moment pour activer et désactiver la connexion. Si défini sur la chaîne "debug", les lignes de résultat seront également imprimées sur la sortie standard. Cet indicateur contrôle finalement un enregistreur Python; voir Configuration de la journalisation pour plus d'informations sur la configuration directe de la journalisation.

Source: Configuration du moteur SQLAlchemy

S'il est utilisé avec Flask, vous pouvez simplement définir

app.config["SQLALCHEMY_ECHO"] = True

pour obtenir le même comportement.

Vedran Šego
la source
6
Cette réponse mérite d'être bien plus élevée .. et pour les utilisateurs de flask-sqlalchemycela devrait être la réponse acceptée.
jso
25

Nous pouvons utiliser la méthode de compilation à cet effet. À partir de la documentation :

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

Résultat:

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

Avertissement de la documentation:

N'utilisez jamais cette technique avec un contenu de chaîne reçu à partir d'une entrée non approuvée, telle que des formulaires Web ou d'autres applications d'entrée utilisateur. Les fonctionnalités de SQLAlchemy pour contraindre les valeurs Python en valeurs de chaîne SQL directes ne sont pas protégées contre les entrées non approuvées et ne valident pas le type de données transmises. Utilisez toujours des paramètres liés lors de l'appel par programme d'instructions SQL non DDL sur une base de données relationnelle.

akshaynagpal
la source
13

Donc, en me basant sur les commentaires de @ zzzeek sur le code de @ bukzor, j'ai trouvé ceci pour obtenir facilement une requête "assez imprimable":

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)

sqlparsePersonnellement, j'ai du mal à lire du code qui n'est pas indenté, j'ai donc l'habitude de réindenter le SQL. Il peut être installé avec pip install sqlparse.

jmagnusson
la source
@bukzor Toutes les valeurs fonctionnent sauf datatime.now()celle lors de l'utilisation de python 3 + sqlalchemy 1.0. Vous devrez suivre les conseils de @zzzeek sur la création d'un TypeDecorator personnalisé pour que celui-ci fonctionne également.
jmagnusson
C'est un peu trop précis. Le datetime ne fonctionne dans aucune combinaison de python et sqlalchemy. De plus, dans py27, l'unicode non-ascii provoque une explosion.
bukzor le
Autant que je puisse voir, la route TypeDecorator m'oblige à modifier mes définitions de table, ce qui n'est pas une exigence raisonnable pour simplement voir mes requêtes. J'ai modifié ma réponse pour qu'elle soit un peu plus proche de la vôtre et de celle de zzzeek, ​​mais j'ai emprunté la voie d'un dialecte personnalisé, qui est correctement orthogonal aux définitions de table.
bukzor le
11

Ce code est basé sur la brillante réponse existante de @bukzor. Je viens d'ajouter un rendu personnalisé pour le datetime.datetimetype dans Oracle TO_DATE().

N'hésitez pas à mettre à jour le code en fonction de votre base de données:

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)
vvladymyrov
la source
22
Je ne vois pas pourquoi les SA pensent qu'il est raisonnable qu'une opération aussi simple soit si difficile .
bukzor
Je vous remercie! render_literal_value a bien fonctionné pour moi. Mon seul changement était: return "%s" % valueau lieu de return repr(value)dans la section float, int, long parce que Python produisait des longs comme 22Lau lieu de juste22
OrganicPanda
Cette recette (ainsi que l'original) déclenche une erreur UnicodeDecodeError si une valeur de chaîne bindparam n'est pas représentable dans ascii. J'ai publié un résumé qui corrige ce problème.
gsakkis
1
"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S")dans mysql
Zitrax
1
@bukzor - Je ne me souviens pas qu'on m'ait demandé si ce qui précède est "raisonnable", donc vous ne pouvez pas vraiment dire que je "crois" que c'est - FWIW, ce n'est pas le cas! :) s'il vous plaît voir ma réponse.
zzzeek
8

Je tiens à souligner que les solutions données ci-dessus ne "fonctionnent pas seulement" avec des requêtes non triviales. Un problème que j'ai rencontré était des types plus compliqués, tels que les ARRAY pgsql causant des problèmes. J'ai trouvé une solution qui, pour moi, fonctionnait même avec les ARRAY pgsql:

emprunté à: https://gist.github.com/gsakkis/4572159

Le code lié semble être basé sur une ancienne version de SQLAlchemy. Vous obtiendrez une erreur indiquant que l'attribut _mapper_zero_or_none n'existe pas. Voici une version mise à jour qui fonctionnera avec une version plus récente, il vous suffit de remplacer _mapper_zero_or_none par bind. De plus, cela prend en charge les tableaux pgsql:

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

Testé à deux niveaux de tableaux imbriqués.

JamesHutchison
la source
Veuillez montrer un exemple de comment l'utiliser? Merci
slashdottir
from file import render_query; print(render_query(query))
Alfonso Pérez
C'est le seul exemple de toute cette page qui a fonctionné pour moi! Merci !
fougerejo