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

102

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!

cce
fonte

Respostas:

106

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.

AndyBarr
fonte
2
Existe uma maneira fácil de obter os valores como um dicionário?
Damien
6
@Damien dado c = q.statement.compile(...), você pode apenas obterc.params
Hannele
1
A postagem está marcada com mysql, então os detalhes do postgresql nesta resposta não são realmente relevantes.
Hannele de
4
Se entendi o OP corretamente, ele quer a consulta final. Imprimir com a especificação de um dialeto (aqui postgres) ainda me dá os espaços reservados em vez dos valores literais. A resposta de @ Matt dá conta do recado. Obter o SQL com espaços reservados pode ser mais simples com o as_scalar()-método de Query.
Patrick B.
1
@PatrickB. Concordo. A resposta de Matt deve ser considerada a resposta "correta". Obtenho o mesmo resultado apenas fazendo str(q).
André C. Andersen
90

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.

Matt
fonte
Obrigado! Isso foi extremamente útil, me permitiu usar a função read_sql do pandas sem dor!
Justin Palmer
24

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)
Albertov
fonte
2
Obrigado por isso! Infelizmente estou usando MySQL, então meu dialeto é "posicional" e precisa ter uma lista de parâmetros em vez de um dicionário. Atualmente tentando fazer seu exemplo funcionar com isso ..
cce
Por favor, não use adaptdesta maneira. No mínimo, chame prepare () no valor de retorno dele a cada vez, fornecendo a conexão como um argumento, para que possa fazer aspas adequadas.
Alex Gaynor
@Alex: Qual seria a maneira correta de fazer uma citação adequada com psycopg? (além de chamar prepare () no valor de retorno, que você parece sugerir que não é o ideal)
albertov
Desculpe, acho que meu fraseado foi ruim, contanto que você chame obj.prepare (conexão), você deve estar bem. Isso ocorre porque APIs "boas" que a libpq fornece para cotação requerem a conexão (e fornece coisas como codificação para strings Unicode).
Alex Gaynor
1
Obrigado. Eu tentei chamada prepareno valor de retorno, mas é parece que ele não tem esse método: AttributeError: 'psycopg2._psycopg.AsIs' object has no attribute 'prepare'. Estou usando o psycopg2 2.2.1 BTW
albertov
18

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)
cce
fonte
Impressionante! Eu só precisava que a lista de parâmetros vinculados fosse enviada ao MySQL e modificar o acima para return tuple(params)funcionar perfeitamente ! Você me poupou incontáveis ​​horas de ter que percorrer um caminho extremamente doloroso.
horcle_buzz
15

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.

nosklo
fonte
porque não é a mesma coisa? Eu entendo que o DB-API está fazendo transações, possivelmente reordenando consultas, etc, mas poderia estar modificando minha consulta mais do que isso?
cce
1
@cce: você está tentando encontrar a consulta final. SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC É a pergunta final. Eles %ssão enviados para o banco de dados por sqlalchemy - sqlalchemy NUNCA coloca os dados reais no lugar de% s
nosklo
@cce: Alguns módulos dbapi também não fazem isso - isso geralmente é feito pelo próprio banco de dados
nosklo
1
aha vejo o que você está dizendo, obrigado - investigando mais a fundo sqlalchemy.dialects.mysql.mysqldb, do_executemany()passa a instrução e os parâmetros separadamente para o cursor MySQLdb. yay indireção!
cce
11

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.

Hannele
fonte
De dentro do depurador PyCharm, o seguinte funcionou para mim ... qry.compile (). Params
Ben
Interessante, pode ser que SQLAlchemy mudou um pouco desde que escrevi esta resposta.
Hannele de
10

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"}}'}
retalógico
fonte
4

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 )
Eric
fonte
Como esta resposta é melhor / diferente da de AndyBarr postada 2 anos antes?
Piotr Dobrogost,
A resposta de AndyBarr inclui um exemplo de geração de uma instrução de consulta com um DBSession, enquanto esta resposta inclui um exemplo usando a API declarativa e o método select. Com relação à compilação da instrução de consulta com um determinado dialeto, as respostas são as mesmas. Eu uso SQLAlchemy para gerar consultas brutas e, em seguida, executá-las com o adbapi do Twister. Para este caso de uso, é útil saber como compilar a consulta sem uma sessão e extrair a string e os parâmetros da consulta.
Eric
3

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

Alex Bender
fonte
2

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())
David K. Hess
fonte
0

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
user2757902
fonte
A instrução não mostra quais são os parâmetros, se você tiver alguns tipos de filtros configurados.
Hannele