SQLAlchemy: imprime la consulta real
Realmente me gustaría poder imprimir SQL válido para mi aplicación, incluidos los valores, en lugar de vincular parámetros, pero no es obvio cómo hacerlo en SQLAlchemy (por diseño, estoy bastante seguro).
¿Alguien ha solucionado este problema de forma general?
En la gran mayoría de los casos, la "stringificación" de una declaración o consulta SQLAlchemy es tan simple como:
print(str(statement))
Esto se aplica tanto a un ORM Querycomo a cualquier select()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í aún no está vinculada a uno, puede pasar esto a compile() :
print(statement.compile(someengine))
o sin motor:
from sqlalchemy.dialects import postgresql
print(statement.compile(dialect=postgresql.dialect()))
Cuando se nos da un objeto ORM Query, para acceder al compile()método 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 vinculados deben estar "incorporados" en la cadena final, el desafío aquí es que SQLAlchemy normalmente no tiene la tarea de esto, ya que esto lo maneja adecuadamente Python DBAPI, sin mencionar que omitir los parámetros vinculados es Probablemente los agujeros de seguridad más explotados en las aplicaciones web modernas. SQLAlchemy tiene una capacidad limitada para realizar esta cadena en determinadas circunstancias, como la emisión de DDL. Para acceder a esta funcionalidad se puede utilizar 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 las advertencias de que solo es compatible con tipos básicos, como ints y cadenas, y además, si bindparam
se usa directamente a sin un valor preestablecido, tampoco podrá encadenarlo.
Para admitir la representación literal en línea para tipos no admitidos, implemente un TypeDecoratorpara el tipo de destino que incluya un
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})
)
produciendo resultados como:
SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)
Dado que lo que desea tiene sentido solo al depurar, puede iniciar SQLAlchemy con echo=Truepara 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– siTrue, el motor registrará todas las declaraciones, así como unarepr()de sus listas de parámetros, en el registrador del motor, cuyo valor predeterminado essys.stdout. Elechoatributo deEnginese puede modificar en cualquier momento para activar y desactivar el inicio de sesión. Si se establece en la cadena"debug", las filas de resultados también se imprimirán en la salida estándar. Esta bandera controla en última instancia un registrador de 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 usa con Flask, simplemente puede configurar
app.config["SQLALCHEMY_ECHO"] = True
para obtener el mismo comportamiento.
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
Podemos utilizar el método de compilació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'
Advertencia de los documentos:
Nunca utilice esta técnica con contenido de cadena recibido de entradas que no sean de confianza, como formularios web u otras aplicaciones de entrada de usuarios. Las funciones de SQLAlchemy para convertir valores de Python en valores de cadena SQL directos no son seguras contra entradas que no son de confianza y no validan el tipo de datos que se pasan. Utilice siempre parámetros vinculados al invocar mediante programación sentencias SQL que no sean DDL en una base de datos relacional.
