Мне бы очень хотелось иметь возможность распечатывать действительный SQL для моего приложения, включая значения, а не привязывать параметры, но неясно, как это сделать в SQLAlchemy (по дизайну, я уверен).
Кто-нибудь решил эту проблему в общем?
SQLAlchemy: распечатать фактический запрос
Ответ 1
Это работает в python 2 и 3 и немного чище, чем раньше, но требует 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
Демо:
# 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()
Дает этот результат: (проверен на python 2.7 и 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
Ответ 2
В подавляющем большинстве случаев "строгая" инструкция или запрос SQLAlchemy так же проста, как:
print str(statement)
Это относится как к ORM Query, так и к любому select() или другому выражению.
Примечание: следующий подробный ответ поддерживается в документации sqlalchemy.
Чтобы получить инструкцию, скомпилированную на конкретный диалект или движок, если сам оператор еще не привязан к одному, вы можете передать это в компилировать():
print statement.compile(someengine)
или без двигателя:
from sqlalchemy.dialects import postgresql
print statement.compile(dialect=postgresql.dialect())
При предоставлении объекта ORM Query, чтобы получить метод compile(), нам нужен только доступ к .statement первый:
statement = query.statement
print statement.compile(someengine)
в отношении первоначальной оговорки о том, что связанные параметры должны быть "встроены" в конечную строку, проблема здесь заключается в том, что SQLAlchemy обычно не задается этим вопросом, так как это правильно обрабатывается с помощью DBAPI Python, не говоря уже об обходе связанные параметры, вероятно, являются наиболее широко используемыми дырами безопасности в современных веб-приложениях. SQLAlchemy имеет ограниченную возможность выполнять эту стрингзацию при определенных обстоятельствах, например, при использовании DDL. Чтобы получить доступ к этой функции, вы можете использовать флаг "literal_binds", переданный в 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})
вышеупомянутый подход имеет оговорки, что он поддерживается только для основных
типы, такие как ints и строки, и, кроме того, если a bindparam
без предварительно заданного значения используется напрямую, он не сможет
это тоже.
Для поддержки встроенного литерального рендеринга для типов, не поддерживаемых, выполните
a TypeDecorator для целевого типа, который включает
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})
)
выводящий вывод как:
SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)
Ответ 3
Учитывая, что то, что вы хотите, имеет смысл только при отладке, вы можете запустить SQLAlchemy с echo=True, чтобы регистрировать все запросы SQL. Например:
engine = create_engine(
"mysql://scott:[email protected]/dbname",
encoding="latin1",
echo=True,
)
Это также может быть изменено только для одного запроса:
echo=False- еслиTrue, Engine регистрирует все операторы, а такжеrepr()их списков параметров в регистраторе движков, по умолчаниюsys.stdout. АтрибутechoEngineможно изменить в любое время, чтобы включить или выключить регистрацию. Если задана строка"debug", строки результатов будут также выводиться на стандартный вывод. Этот флаг в конечном счете управляет регистратором Python; см. Настройка ведения журнала для получения информации о том, как напрямую настроить ведение журнала.Источник: SQLAlchemy Engine Configuration
Если вы используете Flask, вы можете просто установить
app.config["SQLALCHEMY_ECHO"] = True
чтобы получить такое же поведение.
Ответ 4
Мы можем использовать метод компиляции для этой цели. Из документов:
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}))
Результат:
SELECT * FROM users WHERE users.name BETWEEN 'm' AND 'z'
Предупреждение из документов:
Никогда не используйте эту технику с содержимым строки, полученным от ненадежного ввод, например, из веб-форм или других приложений ввода пользователя. Средства SQLAlchemys для преобразования значений Python в прямую строку SQL значения не защищены от ненадежного ввода и не проверяют тип передаваемых данных. Всегда используйте связанные параметры, когда программно вызывать не-DDL операторы SQL против реляционного база данных.
Ответ 5
Итак, построим комментарии @zzzeek по поводу кода @bukzor. Я придумал это, чтобы легко получить запрос "довольно-печатный":
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)
Мне лично трудно прочитать код, который не имеет отступов, поэтому я использовал sqlparse для повторного использования SQL. Его можно установить с помощью pip install sqlparse.
Ответ 6
Этот код основан на блестящем существующем ответе от @bukzor. Я просто добавил пользовательский рендеринг для типа datetime.datetime в Oracle TO_DATE().
Не стесняйтесь обновлять код в соответствии с вашей базой данных:
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)
Ответ 7
Я хотел бы указать, что приведенные выше решения не "просто работают" с нетривиальными запросами. Одна проблема, с которой я столкнулся, - более сложные типы, такие как pgsql ARRAY, вызывающие проблемы. Я нашел решение, которое для меня просто работало даже с pgsql ARRAYs:
заимствован из: https://gist.github.com/gsakkis/4572159
Связанный код, похоже, основан на более старой версии SQLAlchemy. Вы получите сообщение об ошибке, указывающее, что атрибут _mapper_zero_or_none не существует. Здесь обновленная версия, которая будет работать с более новой версией, вы просто замените _mapper_zero_or_none на bind. Кроме того, это поддерживает массивы 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/info/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)
Протестировано до двух уровней вложенных массивов.