SQLAlchemy: imprimir la consulta real

Realmente me gustaría poder imprimir SQL válido para mi aplicación, incluidos los valores, en lugar de los 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 una manera general?

En la gran mayoría de los casos, la “clasificación” de una statement o consulta SQLAlchemy es tan simple como:

 print str(statement) 

Esto se aplica tanto a una Query ORM como a cualquier statement select() u otra.

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

Para obtener la statement comstackda en un dialecto o motor específico, si la statement en sí no está vinculada a una, puede pasarla a compile () :

 print statement.compile(someengine) 

o sin motor:

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

Cuando se proporciona un objeto de Query ORM, para poder acceder al método compile() solo necesitamos acceder primero al descriptor de acceso .statement :

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

con respecto a la estipulación original de que los parámetros enlazados deben estar “incorporados” en la cadena final, el desafío aquí es que SQLAlchemy normalmente no se encarga de esto, ya que esto es manejado apropiadamente por el DBAPI de Python, por no mencionar que eludir los parámetros enlazados es Probablemente los agujeros de seguridad más ampliamente explotados en las aplicaciones web modernas. SQLAlchemy tiene una capacidad limitada para realizar esta clasificación en ciertas circunstancias, como la de emitir DDL. Para acceder a esta funcionalidad, se puede usar el indicador ‘literal_binds’, que se pasa a compile_kwargs :

 from sqlalchemy.sql import table, column, select t = table('t', column('x')) s = select([t]).where(tcx == 5) print s.compile(compile_kwargs={"literal_binds": True}) 

el enfoque anterior tiene la advertencia de que solo se admite para los tipos básicos, como ints y cadenas, y además, si se usa directamente un bindparam sin un valor preestablecido, tampoco podrá hacerlo.

Para admitir la representación literal en línea para tipos no admitidos, 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.cx > 5).compile( compile_kwargs={"literal_binds": True}) ) 

produciendo salida como:

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

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 este resultado: (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 

Entonces, basándose en los comentarios de @zzzeek en 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, por lo que he usado sqlparse para sqlparse a sqlparse el SQL. Se puede instalar con pip install sqlparse .

Dado que lo que desea tiene sentido solo cuando se realiza la depuración, puede iniciar SQLAlchemy con echo=True , para registrar todas las consultas SQL. Por ejemplo:

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

Esto también se puede modificar para una sola solicitud:

echo=False : si es True , el motor registrará todas las declaraciones, así como una repr() de sus listas de parámetros en el registrador de motores, que de forma predeterminada es sys.stdout . El atributo echo de Engine se puede modificar en cualquier momento para activar y desactivar el registro. Si se establece en la cadena "debug" , las filas de resultados también se imprimirán en la salida estándar. Esta bandera finalmente controla un registrador Python; consulte Configuración del registro para obtener información sobre cómo configurar el registro directamente.

Fuente: Configuración del motor SQLAlchemy

Si se utiliza con Flask, simplemente puede configurar

 app.config["SQLALCHEMY_ECHO"] = True 

Para obtener el mismo comportamiento.

Este código se basa en la shiny respuesta existente de @bukzor. Acabo de agregar una representación personalizada para el tipo datetime.datetime en TO_DATE() Oracle.

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) 

Podemos utilizar el método de comstackción para este propósito. De los 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' 

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

prestado 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://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) 

Probado a dos niveles de matrices anidadas.