SQLAlchemy：打印实际查询
Posted
技术标签:
【中文标题】SQLAlchemy：打印实际查询【英文标题】：SQLAlchemy: print the actual query 【发布时间】：2011-08-03 14:48:26 【问题描述】：我真的很希望能够为我的应用程序打印出有效的 SQL，包括值，而不是绑定参数，但是在 SQLAlchemy 中如何做到这一点并不明显（按照设计，我很确定）。
有没有人以一般的方式解决了这个问题？
【问题讨论】：
我没有，但您可以通过利用 SQLAlchemy 的sqlalchemy.engine 日志构建一个不那么脆弱的解决方案。它记录查询和绑定参数，您只需将绑定占位符替换为易于构建的 SQL 查询字符串上的值。
@Simon：使用记录器有两个问题：1）它只在语句执行时打印 2）我仍然需要进行字符串替换，除了在这种情况下，我不会确切知道绑定模板字符串，我必须以某种方式将其从查询文本中解析出来，从而使解决方案更加变得脆弱。
@zzzeek 的常见问题解答的新 URL 似乎是 docs.sqlalchemy.org/en/latest/faq/…。
【参考方案1】：
在绝大多数情况下，SQLAlchemy 语句或查询的“字符串化”很简单：
print(str(statement))
这适用于 ORM Query 以及任何 select() 或其他语句。
注意：以下详细答案正在sqlalchemy documentation 上维护。
要将语句编译为特定方言或引擎，如果语句本身尚未绑定到某个语句，您可以将其传递给compile()：
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 通常不负责此任务，因为这由 Python DBAPI 适当处理，更不用说绕过绑定参数可能是现代 Web 应用程序中利用最广泛的安全漏洞。 SQLAlchemy 在某些情况下（例如发出 DDL）执行此字符串化的能力有限。为了访问此功能，可以使用传递给compile_kwargs 的“literal_binds”标志：
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))
上述方法有一些注意事项，即它仅支持基本的
类型，例如整数和字符串，此外，如果 bindparam
没有预设值直接使用，将无法
将其字符串化。
要支持不支持的类型的内联文字呈现，请实现
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)
【讨论】：
我们暂时停留在 0.7，所以我仍然需要使用我自己的答案。 @zzzeek 为什么默认情况下 sqlalchemy 中不包含漂亮打印查询？喜欢query.prettyprint()。它极大地减轻了大查询的调试痛苦。
@jmagnusson 因为美丽在旁观者的眼中 :) 有足够多的钩子（例如 cursor_execute 事件、Python 日志过滤器、@compiles 等）供任意数量的第三方包实现漂亮的打印系统。
@buzkor re：在 1.0 中修复的限制 bitbucket.org/zzzeek/sqlalchemy/issue/3034/…
对我来说它正在寻找from sqlalchemy.dialects import postgresql; print(query.statement.compile(dialect=postgresql.dialect(), compile_kwargs="literal_binds": True)))【参考方案2】：
鉴于您想要的只有在调试时才有意义，您可以使用echo=True 启动 SQLAlchemy，以记录所有 SQL 查询。例如：
engine = create_engine(
"mysql://scott:tiger@hostname/dbname",
encoding="latin1",
echo=True,
)
这也可以针对单个请求进行修改：
echo=False– 如果True，引擎会将所有语句以及参数列表的repr()记录到引擎记录器，默认为sys.stdout。可以随时修改Engine的echo属性来开启和关闭登录。如果设置为字符串"debug"，结果行也将打印到标准输出。这个标志最终控制了一个 Python 记录器；有关如何直接配置日志记录的信息，请参阅 Configuring Logging。来源：SQLAlchemy Engine Configuration
如果与Flask一起使用，你可以简单地设置
app.config["SQLALCHEMY_ECHO"] = True
获得相同的行为。
【讨论】：
这个答案应该更高......对于flask-sqlalchemy的用户来说，这应该是公认的答案。
是的，这是“正确”的答案。也许它在 2014 年就不存在了？？
@MikeWilliamson，您可以发现它使用了here，这是 2012 年 10 月的文档，于 2013 年 3 月更新。诚然，这比问题（从 2011 年 4 月开始）更新。
【参考方案3】：
这适用于 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
【讨论】：
这太棒了......必须将它添加到一些调试库中，以便我们可以轻松访问它。谢谢你在这个上做步法。我很惊讶它必须如此复杂。 我很确定这很困难，因为新手很想 cursor.execute() 那个字符串。不过，python 中通常使用成年人同意的原则。 确实很好。我冒昧地将其合并到***.com/a/42066590/2127439，其中涵盖了 SQLAlchemy v0.7.9 - v1.1.15，包括 INSERT 和 UPDATE 语句 (PY2/PY3)。 非常好。但它是否转换如下。 1) query(Table).filter(Table.Column1.is_(False) to WHERE Column1 IS 0. 2) query(Table).filter(Table.Column1.is_(True) to WHERE Column1 IS 1. 3) query( Table).filter(Table.Column1 == func.any([1,2,3])) 到 WHERE Column1 = any('[1,2,3]') 上面的转换语法不正确。 如何解决插入语句中的默认值？ ``` python from sqlalchemy import MetaData, insert from sqlalchemy.schema import Column, Table meta = MetaData() TGR = Table('tbl', meta, Column('a'), Column('b'), Column(' c', default='xxx') ) s = TGR.insert().values(a=1, b=2) print(literalquery(s)) ``` INSERT INTO tbl (a, b, c) VALUES ( 1, 2, :c)【参考方案4】：为此，我们可以使用compile 方法。来自docs：
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'
来自文档的警告：
切勿对从不受信任的字符串内容使用此技术 输入，例如来自 Web 表单或其他用户输入应用程序。 SQLAlchemy 将 Python 值强制转换为直接 SQL 字符串的工具 值对于不受信任的输入是不安全的，并且不验证 传递的数据类型。始终使用绑定参数 以编程方式针对关系调用非 DDL SQL 语句 数据库。
【讨论】：
方言请参考本页：docs.sqlalchemy.org/en/13/dialects 这是最好的答案。【参考方案5】：因此，基于@bukzor 代码上的@zzzeek 的cmets，我想出了这个来轻松获得“可打印”的查询：
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安装。
【讨论】：
@bukzor 在使用 python 3 + sqlalchemy 1.0 时，除了datatime.now() 之外的所有值都有效。您必须按照 @zzzeek 的建议创建自定义 TypeDecorator 才能正常工作。
这有点太具体了。日期时间在 python 和 sqlalchemy 的任何组合中都不起作用。此外，在 py27 中，非 ascii unicode 会导致爆炸。
据我所见，TypeDecorator 路由要求我更改表定义，这不是简单查看查询的合理要求。我编辑了我的答案，使其更接近您和 zzzeek 的答案，但我采用了自定义方言的路线，该方言与表定义正确正交。【参考方案6】：
此代码基于来自@bukzor 的出色existing answer。我刚刚将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)
【讨论】：
我不明白为什么 SA 人员认为如此简单的操作如此困难是合理的。 谢谢！ render_literal_value 对我来说效果很好。我唯一的变化是：return "%s" % value 在 float、int、long 部分中而不是 return repr(value)，因为 Python 将 long 输出为 22L 而不仅仅是 22
如果任何 bindparam 字符串值在 ascii 中不可表示，则此配方（以及原始配方）会引发 UnicodeDecodeError。我发布了一个 gist 来解决这个问题。
"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S") 在 mysql 中
@bukzor - 我不记得被问及以上是否“合理”，所以你不能真的说我“相信”它是 - FWIW，它不是！ :) 请看我的回答。【参考方案7】：
我想指出，上面给出的解决方案不能“仅适用于”非平凡的查询。我遇到的一个问题是更复杂的类型，例如导致问题的 pgsql 数组。我确实找到了一个解决方案，对我来说，即使使用 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://***.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)
测试了两层嵌套数组。
【讨论】：
请举例说明如何使用它？谢谢from file import render_query; print(render_query(query))
这是整个页面中唯一对我有用的例子！谢谢！
long 是从哪里导入的？
使用int 而不是long。 long 在 Python 3 中已弃用。【参考方案8】：
只是一个简单的带有 ORM 查询和 pygments 的彩色示例。
import sqlparse
from pygments import highlight
from pygments.formatters.terminal import TerminalFormatter
from pygments.lexers import SqlLexer
from sqlalchemy import create_engine
from sqlalchemy.orm import Query
engine = create_engine("sqlite+pysqlite:///db.sqlite", echo=True, future=True)
def format_sql(query: Query):
compiled = query.statement.compile(
engine, compile_kwargs="literal_binds": True)
parsed = sqlparse.format(str(compiled), reindent=True, keyword_case='upper')
print(highlight(parsed, SqlLexer(), TerminalFormatter()))
或不带sqlparse的版本（不带sqlparse的输出新行少）
def format_sql(query: Query):
compiled = query.statement.compile(
engine, compile_kwargs="literal_binds": True)
print(highlight(str(compiled), SqlLexer(), TerminalFormatter()))
【讨论】：
【参考方案9】：使用 Python 日志记录而不是
echo=True标志来记录 SQL 查询：import logging logging.basicConfig() logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
根据documentation。
【讨论】：
以上是关于SQLAlchemy：打印实际查询的主要内容，如果未能解决你的问题，请参考以下文章