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

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

L'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 chercher dans query._params mais c'est un dict vide. J'ai écrit mon propre compilateur en utilisant cet exemple du sqlalchemy.ext.compiler.compilesdécorateur, mais même la déclaration contient toujours %soù je veux des données.

Je ne peux pas vraiment comprendre quand mes paramètres sont mélangés pour créer la requête; lors de l'examen de l'objet de requête, il s'agit toujours d'un dictionnaire vide (bien que la requête s'exécute correctement et que le moteur l'imprime lorsque vous activez la journalisation d'écho).

Je commence à recevoir le message que SQLAlchemy ne veut pas que je connaisse la requête sous-jacente, car elle rompt la nature générale de l'interface de l'API d'expression avec toutes les différentes DB-API. Cela ne me dérange pas si la requête est exécutée avant que je découvre ce que c'était; Je veux juste savoir!

Ce blog fournit une réponse mise à jour.

Citant l'article du blog, cela est suggéré et 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 n'importe quel type de session.query ().

Merci à Nicolas Cadou pour la réponse! J'espère que cela aidera les autres qui viennent chercher ici.

La documentation utilise literal_bindspour imprimer une requête qcomprenant des paramètres:

print(q.statement.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.

La documentation émet également cet avertissement:

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.

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)

Pour le backend MySQLdb, j'ai un peu modifié la réponse géniale d'albertov (merci beaucoup!). Je suis sûr qu'ils pourraient être fusionnés pour vérifier si comp.positionalc'était le cas , Truemais cela dépasse légèrement la portée 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)

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 produit dans votre base de données.

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

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

Permettez-moi d'abord de dire que je suppose que vous faites cela principalement à des fins de débogage - je ne recommanderais pas d'essayer de modifier l'instruction en dehors de l'API SQLAlchemy fluent.

Malheureusement, il ne semble pas y avoir de moyen simple d'afficher l'instruction compilée avec les paramètres de requête inclus. SQLAlchemy ne met pas réellement les paramètres dans l'instruction - ils sont transmis au moteur de base de données sous forme de dictionnaire . Cela permet à la bibliothèque spécifique à la base de données de gérer des choses comme l'échappement des caractères spéciaux pour éviter l'injection SQL.

Mais vous pouvez le faire en deux étapes assez facilement. Pour obtenir l'instruction, vous pouvez faire comme vous l'avez déjà montré, et simplement imprimer la requête:

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

Vous pouvez faire un pas de plus avec query.statement, pour voir les noms des paramètres. Remarque :id_1ci-dessous vs %sci-dessus - pas vraiment un problème dans cet exemple très simple, mais pourrait être la clé dans une déclaration plus compliquée.

>>> print(query.statement)
>>> print(query.statement.compile()) # seems to be equivalent, you can also
                                     # pass in a dialect if you want
SELECT field_1, field_2 FROM table WHERE id=:id_1;

Ensuite, vous pouvez obtenir les valeurs réelles des paramètres en obtenant la paramspropriété de l'instruction compilée:

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

Cela a fonctionné pour un backend MySQL au moins; Je m'attendrais à ce que ce soit également assez général pour PostgreSQL sans avoir besoin d'utiliser psycopg2.

Pour le backend postgresql utilisant psycopg2, vous pouvez écouter l' do_executeévénement, puis utiliser le curseur, l'instruction et le type de paramètres coercés avec Cursor.mogrify()pour incorporer les paramètres. Vous pouvez renvoyer True pour empêcher l'exécution réelle de la requête.

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

Exemple d'utilisation:

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

La solution suivante utilise le langage d'expression SQLAlchemy et fonctionne avec SQLAlchemy 1.1. Cette solution ne mélange pas les paramètres avec la requête (comme demandé par l'auteur d'origine), mais fournit un moyen d'utiliser les modèles SQLAlchemy pour générer des chaînes de requête SQL et des dictionnaires de paramètres pour différents dialectes SQL. L'exemple est basé sur le tutoriel http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Compte tenu de la classe,

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

nous pouvons produire une instruction de requête en utilisant la fonction de sélection .

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

Ensuite, nous pouvons compiler l'instruction dans un objet de requête.

query = statement.compile()

Par défaut, l'instruction est compilée à l'aide d'une implémentation «nommée» de base compatible avec les bases de données SQL telles que SQLite et Oracle. Si vous devez spécifier un dialecte tel que PostgreSQL, vous pouvez faire

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

Ou si vous souhaitez spécifier explicitement le dialecte en tant que SQLite, vous pouvez changer le style de paramètre de «qmark» à «nommé».

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

À partir de l'objet de requête, nous pouvons extraire la chaîne de requête et les paramètres de requête

query_str = str(query)
query_params = query.params

et enfin exécutez la requête.

conn.execute( query_str, query_params )

Vous pouvez utiliser les événements de la famille ConnectionEvents : after_cursor_executeou before_cursor_execute.

Dans sqlalchemy UsageRecipes by @zzzeek, ​​vous pouvez trouver cet exemple:

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

Ici vous pouvez accéder à votre relevé

Donc, en rassemblant beaucoup de petits morceaux de ces différentes réponses, j'ai trouvé ce dont j'avais besoin: un simple ensemble de code à insérer et de temps en temps mais de manière fiable (c'est-à-dire gère tous les types de données) saisir le SQL compilé exact envoyé à mon Postgres backend en interrogeant simplement la requête elle-même:

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

Je pense que .statement ferait probablement l'affaire: 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