SQLAlchemy:打印实际查询

我确实希望能够为我的应用程序打印有效的SQL，包括值，而不是绑定参数，但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。

有人用一般的方法解决了这个问题吗?

213335 次浏览
小开
最佳答案

这可以在python2和python3中工作，比以前更干净，但需要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
小开

这段代码基于@bukzor的出色现有的回答。我刚刚在Oracle的TO_DATE()中添加了datetime.datetime类型的自定义渲染。

请随时更新代码，以适应您的数据库:

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)
小开

在绝大多数情况下，“stringification"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通常没有这个任务，因为这是由Python DBAPI适当处理的，更不用说绕过绑定参数可能是现代web应用程序中最广泛利用的安全漏洞。SQLAlchemy在某些情况下(如发出DDL)执行这种字符串化的能力有限。为了访问这个功能，可以使用传递给compile_kwargs的' literal_bindings '标志:

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 如果不直接使用预先设置的值，则无法实现 Stringify that .

要支持不支持的类型的内联文字呈现，实现 TypeDecorator用于目标类型，其中包含 TypeDecorator.process_literal_param方法:< / p > 
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)
小开

所以基于@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安装。

小开

我想指出，上面给出的解决方案并不“只适用于”非平凡的查询。我遇到的一个问题是更复杂的类型，比如pgsql数组会导致问题。我确实找到了一个解决方案，对我来说，只是工作，甚至与pgsql数组:

< p >借鉴: https://gist.github.com/gsakkis/4572159 < / p >

链接的代码似乎基于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/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 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语句 数据库。< / p >
小开

假定你想要的只有在调试时才有意义，你可以用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日志记录器;有关如何直接配置日志的信息，请参阅配置日志记录。

来源:SQLAlchemy引擎配置

如果与Flask一起使用，您可以简单地设置

app.config["SQLALCHEMY_ECHO"] = True

得到相同的行为。

小开

只是一个简单的彩色示例与ORM的查询和pydings。

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()))
小开

使用Python日志记录而不是echo=True标记记录SQL查询:

import logging


logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

根据文档。

小开

这就是我的方法

# query is instance of: from sqlalchemy import select
def raw_query(query):
q = str(query.compile())
p = query.compile().params
for k in p.keys():
v = p.get(k)
if isinstance(v, (int, float, complex)):
q = q.replace(f":{k}", f"{v}")
else:
q = q.replace(f":{k}", f"'{v}'")
print(q)

如何使用:

from sqlalchemy import select


select_query = select([
any_model_table.c["id_account"],
any_model_table.c["id_provider"],
any_model_table.c["id_service"],
func.sum(any_model_table.c["items"]).label("items"),
# @eaf
func.date_format(func.now(), "%Y-%m-%d").label("some_date"),
func.date_format(func.now(), "%Y").label("as_year"),
func.date_format(func.now(), "%m").label("as_month"),
func.date_format(func.now(), "%d").label("as_day"),
]).group_by(
any_model_table.c.id_account,
any_model_table.c.id_provider,
any_model_table.c.id_service
).where(
any_model_table.c.id == 5


).where(
func.date_format(any_model_table.c.dt, "%Y-%m-%d") == datetime.utcnow().strftime('%Y-%m-%d')
)


raw_query(select_query)

提交答案