Como obtenho uma consulta SQL compilada bruta de uma expressão SQLAlchemy?

Eu tenho um objeto de consulta SQLAlchemy e desejo obter o texto da instrução SQL compilada, com todos os seus parâmetros vinculados (por exemplo, nenhuma %sou outras variáveis ​​esperando para serem vinculadas pelo compilador de instrução ou mecanismo de dialeto MySQLdb, etc).

Chamar str()a consulta revela algo assim:

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

Tentei procurar em query._params, mas é um dicionário vazio. Eu escrevi meu próprio compilador usando este exemplo do sqlalchemy.ext.compiler.compilesdecorador, mas mesmo a instrução lá ainda tem %sonde eu quero os dados.

Não consigo descobrir quando meus parâmetros se misturam para criar a consulta; ao examinar o objeto de consulta, eles são sempre um dicionário vazio (embora a consulta execute bem e o mecanismo a imprima quando você ativa o log de eco).

Estou começando a receber a mensagem de que SQLAlchemy não quer que eu saiba a consulta subjacente, já que quebra a natureza geral da interface da API de expressão em todos os DB-APIs diferentes. Não me importo se a consulta for executada antes de eu descobrir o que era; Eu só quero saber!

Este blog fornece uma resposta atualizada.

Citando o post do blog, isso é sugerido e funcionou para mim.

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

Onde q é definido como:

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

Ou apenas qualquer tipo de session.query ().

Obrigado a Nicolas Cadou pela resposta! Espero que ajude outras pessoas que vêm pesquisar aqui.

A documentação usa literal_bindspara imprimir uma consulta qincluindo parâmetros:

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

a abordagem acima tem as ressalvas de que só é suportada para tipos básicos, como ints e strings, e, além disso, se um bindparam () sem um valor predefinido for usado diretamente, ele também não será capaz de restringir isso.

A documentação também emite este aviso:

Nunca use essa técnica com conteúdo de string recebido de entrada não confiável, como de formulários da web ou outros aplicativos de entrada do usuário. Os recursos do SQLAlchemy para forçar os valores Python em valores diretos de string SQL não são seguros contra entrada não confiável e não validam o tipo de dados que está sendo passado. Sempre use parâmetros vinculados ao invocar programaticamente instruções SQL não DDL em um banco de dados relacional.

Isso deve funcionar com 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)

Para o back-end do MySQLdb, modifiquei um pouco a resposta incrível de albertov (muito obrigado!). Tenho certeza de que eles poderiam ser mesclados para verificar se comp.positionalestavam, Truemas isso está um pouco além do escopo desta questão.

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)

O fato é que o sqlalchemy nunca mistura os dados com sua consulta. A consulta e os dados são passados ​​separadamente para o driver de banco de dados subjacente - a interpolação de dados acontece em seu banco de dados.

O Sqlalchemy passa a consulta como você viu no str(myquery)banco de dados e os valores irão em uma tupla separada.

Você poderia usar alguma abordagem em que interpola os dados com a consulta você mesmo (como albertov sugerido abaixo), mas isso não é a mesma coisa que o sqlalchemy está executando.

Primeiro, deixe-me começar dizendo que presumo que você esteja fazendo isso principalmente para fins de depuração - eu não recomendaria tentar modificar a instrução fora da API fluente do SQLAlchemy.

Infelizmente, não parece haver uma maneira simples de mostrar a instrução compilada com os parâmetros de consulta incluídos. SQLAlchemy na verdade não coloca os parâmetros na instrução - eles são passados ​​para o mecanismo de banco de dados como um dicionário . Isso permite que a biblioteca específica do banco de dados controle coisas como o escape de caracteres especiais para evitar injeção de SQL.

Mas você pode fazer isso em um processo de duas etapas com razoável facilidade. Para obter o extrato, você pode fazer como já mostrou e apenas imprimir a consulta:

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

Você pode chegar um passo mais perto com query.statement, para ver os nomes dos parâmetros. Observe :id_1abaixo vs %sacima - não é realmente um problema neste exemplo muito simples, mas pode ser a chave em uma declaração mais complicada.

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

Então, você pode obter os valores reais dos parâmetros obtendo a paramspropriedade da instrução compilada:

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

Isso funcionou para um back-end MySQL, pelo menos; Eu esperaria que também fosse geral o suficiente para PostgreSQL sem a necessidade de usar psycopg2.

Para o backend postgresql usando psycopg2, você pode ouvir o do_executeevento e, em seguida, usar o cursor, a instrução e os parâmetros do tipo coagidos junto com Cursor.mogrify()os parâmetros embutidos. Você pode retornar True para evitar a execução real da 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 amostra:

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

A solução a seguir usa SQLAlchemy Expression Language e funciona com SQLAlchemy 1.1. Esta solução não mistura os parâmetros com a consulta (conforme solicitado pelo autor original), mas fornece uma maneira de usar modelos SQLAlchemy para gerar strings de consulta SQL e dicionários de parâmetro para diferentes dialetos SQL. O exemplo é baseado no tutorial http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Dada a aula,

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 produzir uma instrução de consulta usando a função select .

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

Em seguida, podemos compilar a instrução em um objeto de consulta.

query = statement.compile()

Por padrão, a instrução é compilada usando uma implementação 'nomeada' básica que é compatível com bancos de dados SQL como SQLite e Oracle. Se você precisa especificar um dialeto como PostgreSQL, você pode fazer

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

Ou se você quiser especificar explicitamente o dialeto como SQLite, você pode alterar o estilo de parâmetro de 'qmark' para 'nomeado'.

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

Do objeto de consulta, podemos extrair a string de consulta e os parâmetros de consulta

query_str = str(query)
query_params = query.params

e finalmente execute a consulta.

conn.execute( query_str, query_params )

Você pode usar eventos da família ConnectionEvents : after_cursor_executeou before_cursor_execute.

Em sqlalchemy UsageRecipes de @zzzeek você pode encontrar este exemplo:

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

Aqui você pode obter acesso ao seu extrato

Então, juntando um monte de pequenos pedaços dessas diferentes respostas, eu vim com o que eu precisava: um conjunto simples de código para inserir e ocasionalmente, mas de forma confiável (ou seja, lida com todos os tipos de dados), pegue o SQL compilado exato enviado para o meu Back-end do Postgres apenas interrogando a própria consulta:

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

Acho que .statement possivelmente resolveria o problema: 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