python - tables - sqlalchemy models

SQLAlchemy: imprima la consulta real

Question

Realmente me gustaría poder imprimir SQL válido para mi aplicación, incluidos los valores, en lugar de parámetros de enlace, pero no es obvio cómo hacerlo en SQLAlchemy (por diseño, estoy bastante seguro).

¿Alguien ha resuelto este problema de manera general?

Answer 1

Así que basándome en los comentarios de @zzzeek sobre el código de @ bukzor, se me ocurrió esto para obtener fácilmente una consulta "bastante imprimible":

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)

Personalmente, tengo dificultades para leer el código que no está sangrado, así que he usado sqlparse para sqlparse el SQL. Se puede instalar con pip install sqlparse .

Answer 2

En la gran mayoría de los casos, la "cadena" de una declaración o consulta de SQLAlchemy es tan simple como:

print str(statement)

Esto aplica tanto a una Query ORM como a cualquier select() u otra declaración.

Nota : la siguiente respuesta detallada se mantiene en la documentación de sqlalchemy .

Para obtener la declaración compilada en un dialecto o motor específico, si la declaración en sí misma no está vinculada a uno, puede pasar esto a compile() :

print statement.compile(someengine)

o sin un motor:

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

Cuando se le da un objeto ORM Query , para obtener el método compile() , solo necesitamos acceder .statement al .statement :

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

con respecto a la estipulación original de que los parámetros vinculados deben ser "incorporados" en la cadena final, el desafío aquí es que normalmente no se le asigna a SQLAlchemy, ya que esto es manejado apropiadamente por el DBAPI de Python, sin mencionar que se omiten los parámetros vinculados. probablemente los agujeros de seguridad más explotados en aplicaciones web modernas. SQLAlchemy tiene una capacidad limitada para realizar esta secuencia en ciertas circunstancias, como la de emitir DDL. Para acceder a esta funcionalidad, se puede usar el indicador ''literal_binds'', pasado a 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})

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.

Para admitir la representación literal en línea para tipos no compatibles, implemente un TypeDecorator para el tipo de destino que incluye un método TypeDecorator.process_literal_param :

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

produciendo resultados como:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)

Answer 3

Este código se basa en la brillante respuesta existente de @bukzor. Acabo de agregar un renderizado personalizado para datetime.datetime type en Oracle TO_DATE() .

Siéntase libre de actualizar el código para adaptarse a su base de datos:

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)

Answer 4

Esto funciona en python 2 y 3 y es un poco más limpio que antes, pero requiere 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

Manifestación:

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

Da esta salida: (probado en python 2.7 y 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

Answer 5

Me gustaría señalar que las soluciones dadas anteriormente no "simplemente funcionan" con consultas no triviales. Un problema con el que me encontré fueron los tipos más complicados, como los ARRAY pgsql que causan problemas. Encontré una solución que, para mí, funcionó incluso con los ARRAY pgsql:

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

El código vinculado parece estar basado en una versión anterior de SQLAlchemy. Recibirá un error que dice que el atributo _mapper_zero_or_none no existe. Aquí hay una versión actualizada que funcionará con una versión más nueva, simplemente reemplaza _mapper_zero_or_none con bind. Además, esto tiene soporte para matrices 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://.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)

Probado en dos niveles de matrices anidadas.

Answer 6

Podemos usar el método de compile() para este propósito. De los docs.sqlalchemy.org/en/latest/faq/… :

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