Comment obtenir une requête SQL brute et compilée à partir d'une expression SQLAlchemy ?

Question

J'ai un objet de requête SQLAlchemy et je veux obtenir le texte de la requête SQL compilée, avec tous ses paramètres liés (par exemple, pas de %s ou d'autres variables attendant d'être liées par le compilateur d'instructions ou le moteur de dialecte MySQLdb, etc).

Appel str() sur la requête révèle quelque chose comme ceci :

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

J'ai essayé de regarder dans query._params mais c'est un dict vide. J'ai écrit mon propre compilateur en utilisant cet exemple de la sqlalchemy.ext.compiler.compiles décorateur mais même la déclaration qui s'y trouve a encore %s où je souhaite obtenir des données.

Je n'arrive pas à comprendre quand mes paramètres sont mélangés pour créer la requête ; lorsque j'examine l'objet requête, ils sont toujours un dictionnaire vide (bien que la requête s'exécute correctement et que le moteur l'imprime lorsque vous activez la journalisation de l'écho).

Je commence à comprendre que SQLAlchemy ne veut pas que je connaisse la requête sous-jacente, car cela rompt la nature générale de l'interface de l'API d'expression avec toutes les différentes API de base de données. Je me fiche que la requête soit exécutée avant que je ne sache ce qu'elle était ; je veux juste savoir !

Answer 1

Les la documentation utilise literal_binds pour imprimer une requête q y compris les paramètres :

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

l'approche ci-dessus a pour inconvénient de n'être supportée que pour les types de base, tels que les ints et les chaînes, et en outre, si un bindparam() sans valeur prédéfinie est utilisé directement, il ne sera pas non plus possible d'en faire une chaîne.

La documentation émet également cet avertissement :

N'utilisez jamais cette technique avec du contenu de chaîne reçu d'un comme des formulaires web ou d'autres entrées utilisateur. Les fonctions de SQLAlchemy permettant de forcer les valeurs Python à devenir des valeurs de chaîne SQL directes sont les suivantes ne sont pas sécurisées contre les entrées non fiables et ne permettent pas de valider le type de données transmises. [ ] programmation d'instructions SQL non DDL sur une base de données relationnelle. relationnelle.

Answer 2

Ce billet de Nicolas Cadou fournit une réponse actualisée.

En citant l'article du blog, voici ce qui est suggéré et qui a fonctionné pour moi :

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

Où q est défini comme :

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

Ou tout autre type de session.query() .

Answer 3

Cela devrait fonctionner avec 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 4

Le fait est que sqlalchemy ne mélange jamais les données avec votre requête. La requête et les données sont transmises séparément à votre pilote de base de données sous-jacent - l'interpolation des données se fait dans votre base de données.

Sqlalchemy passe la requête comme vous l'avez vu dans str(myquery) à la base de données, et les valeurs seront placées dans un tuple séparé.

Vous pourriez utiliser une approche où vous interpolez les données avec la requête vous-même (comme albertov l'a suggéré ci-dessous), mais ce n'est pas la même chose que sqlalchemy exécute.

Answer 5

Pour le backend MySQLdb, j'ai un peu modifié l'excellente réponse d'albertov (merci beaucoup !). Je suis sûr qu'ils pourraient être fusionnés pour vérifier si comp.positional était True mais cela dépasse légèrement le cadre de cette question.

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)