python - raw - ¿Cómo obtengo una consulta SQL compilada sin procesar desde una expresión SQLAlchemy?

sqlalchemy tutorial

Question

Tengo un objeto de consulta SQLAlchemy y quiero obtener el texto de la declaración SQL compilada, con todos sus parámetros encuadernados (por ejemplo, no %s u otras variables que esperan ser vinculadas por el compilador de sentencias o el motor de dialecto MySQLdb, etc.).

Llamar a str() en la consulta revela algo como esto:

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

Intenté buscar en query._params pero es un dict vacío. Escribí mi propio compilador usando este ejemplo del decorador sqlalchemy.ext.compiler.compiles pero incluso la declaración todavía tiene %s donde quiero datos.

No puedo entender cuándo se mezclan mis parámetros para crear la consulta; al examinar el objeto de consulta, siempre son un diccionario vacío (aunque la consulta se ejecuta correctamente y el motor la imprime cuando activa el registro de eco).

Estoy comenzando a recibir el mensaje de que SQLAlchemy no quiere que yo conozca la consulta subyacente, ya que rompe la naturaleza general de la interfaz de la API de expresión con todas las diferentes API de DB. No me importa si la consulta se ejecuta antes de descubrir lo que era; ¡Sólo quiero saber!

Answer 1

Creo que .statement posiblemente sería el truco: 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

Answer 2

Esto debería funcionar con 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 3

La documentation usa literal_binds para imprimir una consulta q incluye parámetros:

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

el enfoque anterior tiene la advertencia de que solo es compatible con tipos básicos, como enteros y cadenas, y además, si un bindparam () sin un valor preestablecido se usa directamente, tampoco podrá codificarlo.

Answer 4

La cosa es que sqlalchemy nunca mezcla los datos con su consulta. La consulta y los datos se pasan por separado a su controlador de base de datos subyacente: la interpolación de datos ocurre en su base de datos.

Sqlalchemy pasa la consulta como lo ha visto en str(myquery) a la base de datos, y los valores irán en una tupla separada.

Podría usar algún enfoque en el que interpole los datos con la consulta usted mismo (como albertov se sugiere a continuación), pero eso no es lo mismo que se está ejecutando sqlalchemy.

Answer 5

La siguiente solución usa el lenguaje de expresiones SQLAlchemy y funciona con SQLAlchemy 1.1. Esta solución no mezcla los parámetros con la consulta (según lo solicitado por el autor original), pero proporciona una forma de utilizar los modelos SQLAlchemy para generar cadenas de consulta SQL y diccionarios de parámetros para diferentes dialectos SQL. El ejemplo se basa en el tutorial http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Dada la clase,

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

podemos producir una declaración de consulta usando la función de selección .

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

A continuación, podemos compilar la declaración en un objeto de consulta.

query = statement.compile()

De forma predeterminada, la declaración se compila utilizando una implementación básica ''nombrada'' que es compatible con bases de datos SQL como SQLite y Oracle. Si necesita especificar un dialecto como PostgreSQL, puede hacer

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

O si desea especificar explícitamente el dialecto como SQLite, puede cambiar el paramstyle de ''qmark'' a ''named''.

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

Desde el objeto de consulta, podemos extraer la cadena de consulta y los parámetros de consulta

query_str = str(query)
query_params = query.params

y finalmente ejecuta la consulta.

conn.execute( query_str, query_params )

Answer 6

Para el back-end postgresql usando psycopg2, puede escuchar el evento do_execute , luego usar el cursor, la instrucción y el tipo de parámetros forzados junto con Cursor.mogrify() para Cursor.mogrify() los parámetros. Puede devolver True para evitar la ejecución real de la consulta.

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

Uso de muestra:

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

Para el backend de MySQLdb modifiqué un poco la increíble respuesta de albertov (¡muchas gracias!). Estoy seguro de que podrían fusionarse para comprobar si comp.positional era verdadero, pero eso está un poco fuera del alcance de esta pregunta.

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)

Answer 8

Primero déjame prólogo diciendo que supongo que estás haciendo esto principalmente para depuración: no recomendaría intentar modificar la declaración fuera de la API fluida de SQLAlchemy.

Desafortunadamente, no parece haber una manera simple de mostrar la declaración compilada con los parámetros de consulta incluidos. SQLAlchemy en realidad no pone los parámetros en la declaración; se pasan al motor de la base de datos como un diccionario . Esto permite que el manejador de la biblioteca específica de la base de datos escape caracteres especiales y similares, para evitar la inyección de SQL.

Pero puede hacer esto en un proceso de dos pasos con bastante facilidad. Para obtener la declaración, puede hacer lo que ya ha mostrado y simplemente imprimir la consulta:

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

Puede acercarse un paso más con query.statement, para ver los nombres de los parámetros. (Nota :id_1 debajo vs %s arriba - no es realmente un problema en este ejemplo muy simple, pero podría ser clave en una declaración más complicada.)

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

Luego, puede obtener los valores reales de los parámetros obteniendo la propiedad params de la declaración compilada:

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

Esto funcionó para un backend de MySQL al menos; Yo esperaría que también sea lo suficientemente general para PostgreSQL sin necesidad de usar psycopg2 .

Answer 9

Puede usar eventos de la familia ConnectionEvents : after_cursor_execute o before_cursor_execute .

En sqlalchemy UsageRecipes por @zzzeek puedes encontrar este ejemplo:

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

Aquí puede obtener acceso a su extracto

Answer 10

This blog proporciona una respuesta actualizada.

Citando de la publicación del blog, esto es sugerido y funcionó para mí.

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

Donde q se define como:

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

O simplemente cualquier tipo de sesión.query ().

¡Gracias a Nicolas Cadou por la respuesta! Espero que ayude a otros que vienen buscando aquí.