我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
当前回答
假定只有在调试时才有意义,可以使用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 Engine Configuration
如果与Flask一起使用,您可以简单地设置
app.config["SQLALCHEMY_ECHO"] = True
得到相同的行为。
其他回答
这段代码是基于@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)
使用Python日志代替echo=True标记记录SQL查询: 导入日志 logging.basicConfig () logging.getLogger (sqlalchemy.engine) .setLevel (logging.INFO)
根据文档。
为此,我们可以使用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语句 数据库。
这就是我的方法
# 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)
只是一个简单的彩色示例与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()))