SQLAlchemy: imprime a consulta real

165

Eu realmente gostaria de poder imprimir SQL válido para o meu aplicativo, incluindo valores, em vez de parâmetros de ligação, mas não é óbvio como fazer isso no SQLAlchemy (por design, tenho certeza).

Alguém resolveu esse problema de uma maneira geral?

Bukzor
fonte
1
Não, mas você provavelmente poderia criar uma solução menos frágil tocando no sqlalchemy.enginelog do SQLAlchemy . Ele registra consultas e parâmetros de ligação, você apenas precisará substituir os espaços reservados de ligação pelos valores em uma sequência de consulta SQL prontamente construída.
Simon
@ Simon: há dois problemas com o uso do logger: 1) ele é impresso apenas quando uma instrução está sendo executada 2) Eu ainda teria que substituir uma string, exceto nesse caso, eu não saberia exatamente a string do modelo de ligação , e eu precisaria analisá-lo de alguma forma no texto da consulta, tornando a solução mais frágil.
bukzor
O novo URL parece ser docs.sqlalchemy.org/en/latest/faq/… para as perguntas frequentes do @ zzzeek.
Jim DeLaHunt

Respostas:

167

Na grande maioria dos casos, a "stringification" de uma instrução ou consulta SQLAlchemy é tão simples quanto:

print str(statement)

Isso se aplica tanto a um ORM Queryquanto a qualquer select()outra declaração.

Nota : a resposta detalhada a seguir está sendo mantida na documentação do sqlalchemy .

Para obter a instrução compilada em um dialeto ou mecanismo específico, se a própria declaração ainda não estiver vinculada a uma, você pode passar para compile () :

print statement.compile(someengine)

ou sem um motor:

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

Quando é fornecido um Queryobjeto ORM , para obter o compile()método, precisamos acessar apenas o acessador .statement primeiro:

statement = query.statement
print statement.compile(someengine)

No que diz respeito à estipulação original de que os parâmetros vinculados devem ser "incorporados" na sequência final, o desafio aqui é que o SQLAlchemy normalmente não é encarregado disso, pois isso é tratado adequadamente pelo DBAPI do Python, sem mencionar que os parâmetros vinculados são ignorados. provavelmente as falhas de segurança mais amplamente exploradas em aplicativos da web modernos. O SQLAlchemy possui capacidade limitada para fazer essa stringificação em determinadas circunstâncias, como a de emitir DDL. Para acessar essa funcionalidade, pode-se usar o sinalizador 'literal_binds', passado para compile_kwargs:

from sqlalchemy.sql import table, column, select

t = table('t', column('x'))

s = select([t]).where(t.c.x == 5)

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

a abordagem acima tem as ressalvas de que ele é suportado apenas para tipos básicos, como ints e strings, e além disso, se um valor bindparam sem um valor pré-definido for usado diretamente, também não será possível especificá-lo.

Para oferecer suporte à renderização literal embutida para tipos não suportados, implemente a TypeDecoratorpara o tipo de destino que inclui um TypeDecorator.process_literal_parammétodo:

from sqlalchemy import TypeDecorator, Integer


class MyFancyType(TypeDecorator):
    impl = Integer

    def process_literal_param(self, value, dialect):
        return "my_fancy_formatting(%s)" % value

from sqlalchemy import Table, Column, MetaData

tab = Table('mytable', MetaData(), Column('x', MyFancyType()))

print(
    tab.select().where(tab.c.x > 5).compile(
        compile_kwargs={"literal_binds": True})
)

produzindo saída como:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)
zzzeek
fonte
2
Isso não coloca aspas nas seqüências de caracteres e não resolve alguns parâmetros vinculados.
bukzor
1
a segunda metade da resposta foi atualizada com as informações mais recentes.
Zzzeek
2
@zzzeek Por que as consultas de impressão bonita não estão incluídas no sqlalchemy por padrão? Like query.prettyprint(). Facilita imensamente a dor da depuração com grandes consultas.
precisa saber é o seguinte
2
@jmagnusson porque a beleza está nos olhos de quem vê :) Existem muitos ganchos (por exemplo, evento cursor_execute, filtros de log Python @compiles, etc.) para qualquer número de pacotes de terceiros para implementar sistemas de impressão bonita.
Zzzeek
1
@buzkor re: limit que foi corrigido em 1.0 bitbucket.org/zzzeek/sqlalchemy/issue/3034/…
zzzeek
66

Isso funciona em python 2 e 3 e é um pouco mais limpo do que antes, mas requer SA> = 1.0.

from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType

# python2/3 compatible.
PY3 = str is not bytes
text = str if PY3 else unicode
int_type = int if PY3 else (int, long)
str_type = str if PY3 else (str, unicode)


class StringLiteral(String):
    """Teach SA how to literalize various things."""
    def literal_processor(self, dialect):
        super_processor = super(StringLiteral, self).literal_processor(dialect)

        def process(value):
            if isinstance(value, int_type):
                return text(value)
            if not isinstance(value, str_type):
                value = text(value)
            result = super_processor(value)
            if isinstance(result, bytes):
                result = result.decode(dialect.encoding)
            return result
        return process


class LiteralDialect(DefaultDialect):
    colspecs = {
        # prevent various encoding explosions
        String: StringLiteral,
        # teach SA about how to literalize a datetime
        DateTime: StringLiteral,
        # don't format py2 long integers to NULL
        NullType: StringLiteral,
    }


def literalquery(statement):
    """NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        statement = statement.statement
    return statement.compile(
        dialect=LiteralDialect(),
        compile_kwargs={'literal_binds': True},
    ).string

Demo:

# coding: UTF-8
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        10 ** 20,  # a long integer
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print(literalquery(statement))


if __name__ == '__main__':
    test()

Dá esta saída: (testado em python 2.7 e 3.4)

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
      '2015-06-24 18:09:29.042517', 3.14159, 100000000000000000000)
 LIMIT 1
Bukzor
fonte
2
Isso é incrível ... Teremos que adicionar isso a algumas bibliotecas de depuração para que possamos acessá-lo facilmente. Obrigado por fazer o trabalho de pés neste. Estou surpreso que tivesse que ser tão complicado.
Corey O.
5
Tenho certeza de que isso é intencionalmente difícil, porque os novatos são tentados a cursor.execute () essa string. O princípio de consentir adultos é comumente usado em python.
bukzor
Muito útil. Obrigado!
clime
Muito bom mesmo. Tomei a liberdade e incorporei isso no stackoverflow.com/a/42066590/2127439 , que abrange o SQLAlchemy v0.7.9 - v1.1.15, incluindo instruções INSERT e UPDATE (PY2 / PY3).
wolfmanx
muito agradável. mas está convertendo como abaixo. 1) query (Table) .filter (Table.Column1.is_ (False) para WHERE Column1 IS 0. 2) query (Table) .filter (Table.Column1.is_ (True) para WHERE Column1 IS 1. 3) query ( Table) .filter (Table.Column1 == func.any ([1,2,3])) para WHERE Coluna1 = qualquer ('[1,2,3]') acima das conversões está incorreta na sintaxe.
Sekhar C
51

Dado que o que você deseja faz sentido apenas durante a depuração, você pode iniciar o SQLAlchemy com echo=True, para registrar todas as consultas SQL. Por exemplo:

engine = create_engine(
    "mysql://scott:tiger@hostname/dbname",
    encoding="latin1",
    echo=True,
)

Isso também pode ser modificado para apenas uma única solicitação:

echo=False- se True, o mecanismo registrará todas as instruções e uma repr()de suas listas de parâmetros no criador de logs de mecanismos, cujo padrão é sys.stdout. O echoatributo de Enginepode ser modificado a qualquer momento para ativar e desativar o log. Se definido como a sequência "debug", as linhas de resultado também serão impressas na saída padrão. Esse sinalizador controla o logger do Python; consulte Configurando o log para obter informações sobre como configurar o log diretamente.

Fonte: configuração do mecanismo SQLAlchemy

Se usado com o Flask, você pode simplesmente definir

app.config["SQLALCHEMY_ECHO"] = True

para obter o mesmo comportamento.

Vedran Šego
fonte
6
Essa resposta merece ser muito mais alta e, para os usuários, flask-sqlalchemydeve ser a resposta aceita.
JSO
25

Podemos usar o método de compilação para esse fim. Dos documentos :

from sqlalchemy.sql import text
from sqlalchemy.dialects import postgresql

stmt = text("SELECT * FROM users WHERE users.name BETWEEN :x AND :y")
stmt = stmt.bindparams(x="m", y="z")

print(stmt.compile(dialect=postgresql.dialect(),compile_kwargs={"literal_binds": True}))

Resultado:

SELECT * FROM users WHERE users.name BETWEEN 'm' AND 'z'

Aviso dos documentos:

Nunca use essa técnica com conteúdo de sequência de caracteres recebido de entrada não confiável, como formulários da Web ou outros aplicativos de entrada do usuário. As instalações do SQLAlchemy para coagir valores Python em valores diretos de string SQL não são seguras contra entradas não confiáveis ​​e não validam o tipo de dados que estão sendo transmitidos. Sempre use parâmetros associados ao chamar programaticamente instruções SQL não DDL em um banco de dados relacional.

akshaynagpal
fonte
13

Então, com base nos comentários de @ zzzeek no código de @ bukzor, eu vim com isso para obter facilmente uma consulta "bastante imprimível":

def prettyprintable(statement, dialect=None, reindent=True):
    """Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement. The function can also receive a
    `sqlalchemy.orm.Query` object instead of statement.
    can 

    WARNING: Should only be used for debugging. Inlining parameters is not
             safe when handling user created data.
    """
    import sqlparse
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if dialect is None:
            dialect = statement.session.get_bind().dialect
        statement = statement.statement
    compiled = statement.compile(dialect=dialect,
                                 compile_kwargs={'literal_binds': True})
    return sqlparse.format(str(compiled), reindent=reindent)

Pessoalmente, tenho dificuldade em ler o código que não é recuado, então eu costumava sqlparsereindentar o SQL. Pode ser instalado com pip install sqlparse.

jmagnusson
fonte
@bukzor Todos os valores funcionam, exceto datatime.now()aquele ao usar python 3 + sqlalchemy 1.0. Você teria que seguir o conselho do @ zzzeek sobre como criar um TypeDecorator personalizado para que ele funcionasse também.
Jmagnusson
Isso é um pouco específico demais. O datetime não funciona em nenhuma combinação de python e sqlalchemy. Além disso, na py27, o unicode não-ascii causa uma explosão.
bukzor
Tanto quanto pude ver, a rota TypeDecorator exige que eu altere minhas definições de tabela, o que não é um requisito razoável para simplesmente ver minhas consultas. Editei minha resposta para ficar um pouco mais próxima da sua e do zzzeek, ​​mas segui o caminho de um dialeto personalizado, que é corretamente ortogonal às definições da tabela.
bukzor
11

Esse código é baseado na resposta brilhante existente de @bukzor. Acabei de adicionar renderização personalizada para o datetime.datetimetipo no Oracle TO_DATE().

Sinta-se livre para atualizar o código para se adequar ao seu banco de dados:

import decimal
import datetime

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if bind is None:
            bind = statement.session.get_bind(
                    statement._mapper_zero_or_none()
            )
        statement = statement.statement
    elif bind is None:
        bind = statement.bind 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )
        def render_literal_value(self, value, type_):
            """Render the value of a bind parameter as a quoted literal.

            This is used for statement sections that do not accept bind paramters
            on the target driver/database.

            This should be implemented by subclasses using the quoting services
            of the DBAPI.

            """
            if isinstance(value, basestring):
                value = value.replace("'", "''")
                return "'%s'" % value
            elif value is None:
                return "NULL"
            elif isinstance(value, (float, int, long)):
                return repr(value)
            elif isinstance(value, decimal.Decimal):
                return str(value)
            elif isinstance(value, datetime.datetime):
                return "TO_DATE('%s','YYYY-MM-DD HH24:MI:SS')" % value.strftime("%Y-%m-%d %H:%M:%S")

            else:
                raise NotImplementedError(
                            "Don't know how to literal-quote value %r" % value)            

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)
vvladymyrov
fonte
22
Não vejo por que o pessoal da SA acredita que é razoável que uma operação tão simples seja tão difícil .
bukzor
Obrigado! render_literal_value funcionou bem para mim. Minha única mudança foi: return "%s" % valueem vez de return repr(value)no float, int, longa seção porque Python foi a saída de longs como 22Lem vez de apenas22
OrganicPanda
Esta receita (assim como a original) gera UnicodeDecodeError se algum valor da string bindparam não for representável em ascii. Eu postei uma essência que corrige isso.
precisa saber é o seguinte
1
"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S")no mysql
Zitrax
1
@bukzor - Não me lembro de ter sido perguntado se o acima é "razoável", então você não pode realmente afirmar que "acredito" que é - FWIW, não é! :) por favor, veja minha resposta.
Zzzeek 23/05
8

Gostaria de salientar que as soluções fornecidas acima não "apenas funcionam" com consultas não triviais. Um problema que encontrei foram tipos mais complicados, como ARRAYs do pgsql, causando problemas. Eu encontrei uma solução que, para mim, funcionou mesmo com ARRAYs do pgsql:

emprestado de: https://gist.github.com/gsakkis/4572159

O código vinculado parece basear-se em uma versão mais antiga do SQLAlchemy. Você receberá um erro dizendo que o atributo _mapper_zero_or_none não existe. Aqui está uma versão atualizada que funcionará com uma versão mais recente; você simplesmente substitui _mapper_zero_or_none por bind. Além disso, isso tem suporte para matrizes pgsql:

# adapted from:
# https://gist.github.com/gsakkis/4572159
from datetime import date, timedelta
from datetime import datetime

from sqlalchemy.orm import Query


try:
    basestring
except NameError:
    basestring = str


def render_query(statement, dialect=None):
    """
    Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement.
    WARNING: This method of escaping is insecure, incomplete, and for debugging
    purposes only. Executing SQL statements with inline-rendered user values is
    extremely insecure.
    Based on http://stackoverflow.com/questions/5631078/sqlalchemy-print-the-actual-query
    """
    if isinstance(statement, Query):
        if dialect is None:
            dialect = statement.session.bind.dialect
        statement = statement.statement
    elif dialect is None:
        dialect = statement.bind.dialect

    class LiteralCompiler(dialect.statement_compiler):

        def visit_bindparam(self, bindparam, within_columns_clause=False,
                            literal_binds=False, **kwargs):
            return self.render_literal_value(bindparam.value, bindparam.type)

        def render_array_value(self, val, item_type):
            if isinstance(val, list):
                return "{%s}" % ",".join([self.render_array_value(x, item_type) for x in val])
            return self.render_literal_value(val, item_type)

        def render_literal_value(self, value, type_):
            if isinstance(value, long):
                return str(value)
            elif isinstance(value, (basestring, date, datetime, timedelta)):
                return "'%s'" % str(value).replace("'", "''")
            elif isinstance(value, list):
                return "'{%s}'" % (",".join([self.render_array_value(x, type_.item_type) for x in value]))
            return super(LiteralCompiler, self).render_literal_value(value, type_)

    return LiteralCompiler(dialect, statement).process(statement)

Testado para dois níveis de matrizes aninhadas.

JamesHutchison
fonte
Por favor, mostre um exemplo de como usá-lo? Obrigado
slashdottir
from file import render_query; print(render_query(query))
Alfonso Pérez
Esse é o único exemplo desta página inteira que funcionou para mim! Obrigado !
fougerejo 8/01