我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
当前回答
所以基于@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安装它。
其他回答
为此,我们可以使用compile方法。从文档中可以看出:
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语句 数据库。
只是一个简单的彩色示例与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()))
我想指出,上面给出的解决方案并不“只适用于”非平凡的查询。我遇到的一个问题是更复杂的类型,比如pgsql数组会导致问题。我确实找到了一个解决方案,对我来说,只是工作,甚至与pgsql数组:
借鉴: 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/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)
测试了两层嵌套数组。
使用Python日志代替echo=True标记记录SQL查询: 导入日志 logging.basicConfig () logging.getLogger (sqlalchemy.engine) .setLevel (logging.INFO)
根据文档。
所以基于@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安装它。