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

Tout d'abord, permettez-moi de commencer en disant que je suppose que vous faites cela principalement à des fins de débogage - je ne recommanderais pas d'essayer de modifier la déclaration en dehors de l'API fluide de SQLAlchemy.

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

Mais vous pouvez le faire en deux étapes assez facilement. Pour obtenir la déclaration, vous pouvez faire ce que vous 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_1 ci-dessous vs %s ci-dessus - ce n'est pas vraiment un problème dans cet exemple très simple, mais cela pourrait être une 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 le paramètre params de la déclaration compilée :

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

Cela fonctionne au moins pour un backend MySQL ; je pense que c'est aussi assez général pour PostgreSQL sans avoir besoin d'utiliser psycopg2 .

Answer 2

Pour le backend postgresql utilisant psycopg2, vous pouvez écouter la commande do_execute puis utiliser le curseur, l'instruction et les paramètres coercitifs de type avec le Cursor.mogrify() pour intégrer les paramètres. Vous pouvez renvoyer True pour empêcher l'exécution 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"}}'}

Answer 3

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 le demandait l'auteur original), 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

La classe est donnée,

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 requête à l'aide de la fonction sélectionner fonction.

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

Ensuite, nous pouvons compiler la déclaration en un objet de requête.

query = statement.compile()

Par défaut, la déclaration 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 SQLite, vous pouvez changer le paramstyle de 'qmark' à 'named'.

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

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

query_str = str(query)
query_params = query.params

et enfin exécuter la requête.

conn.execute( query_str, query_params )

Answer 4

Vous pouvez utiliser les événements de Événements de connexion famille : after_cursor_execute o before_cursor_execute .

Dans sqlalchemy UtilisationRecettes par @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 déclaration

Answer 5

MISE À JOUR : J'ai trouvé un autre cas où la solution précédente ne produisait pas correctement la bonne instruction SQL. Après un peu de plongée dans SQLAlchemy, il devient évident que vous ne devez pas seulement compiler pour un dialecte particulier, vous devez aussi prendre la requête compilée et l'initialiser pour le contexte de connexion DBAPI correct. Sinon, des choses comme les processeurs de type bind ne sont pas exécutées et des valeurs comme JSON.NULL ne sont pas traduites correctement.

Attention, cela rend cette solution très particulière à Flask + Flask-SQLAlchemy + psycopg2 + PostgreSQL. Il se peut que vous deviez adapter cette solution à votre environnement en changeant le dialecte et la façon dont vous référencez votre connexion. Cependant, je suis assez confiant dans le fait que cette solution produit le SQL exact pour tous les types de données.

Le résultat ci-dessous est une méthode simple qui permet d'obtenir occasionnellement, mais de manière fiable, le code SQL exact compilé qui serait envoyé à mon backend PostgreSQL en interrogeant simplement la requête elle-même :

import sqlalchemy.dialects.postgresql.psycopg2

from flask import current_app

def query_to_string(query):
    dialect = sqlalchemy.dialects.postgresql.psycopg2.dialect()
    compiled_query = query.statement.compile(dialect=dialect)
    sqlalchemy_connection = current_app.db.session.connection()
    context = dialect.execution_ctx_cls._init_compiled(
        dialect,
        sqlalchemy_connection,
        sqlalchemy_connection.connection,
        compiled_query,
        None
    )
    mogrified_query = sqlalchemy_connection.connection.cursor().mogrify(
        context.statement,
        context.parameters[0]
    )
    return mogrified_query.decode()

query = [ .... some ORM query .... ]

print(f"compiled SQL = {query_to_string(query)}")