我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
当前回答
这就是我的方法
# 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)
其他回答
使用Python日志代替echo=True标记记录SQL查询: 导入日志 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)
这可以在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
只是一个简单的彩色示例与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)
测试了两层嵌套数组。