我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。
有人用一般的方法解决了这个问题吗?
当前回答
这可以在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
其他回答
在绝大多数情况下,SQLAlchemy语句或查询的“字符串化”非常简单:
print(str(statement))
这既适用于ORM查询,也适用于任何select()或其他语句。
注意:下面的详细答案在sqlalchemy文档中维护。
要获得编译为特定方言或引擎的语句,如果语句本身还没有绑定到某个方言或引擎,可以将其传入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))
with regards to the original stipulation that bound parameters are to be "inlined" into the final string, the challenge here is that SQLAlchemy normally is not tasked with this, as this is handled appropriately by the Python DBAPI, not to mention bypassing bound parameters is probably the most widely exploited security holes in modern web applications. SQLAlchemy has limited ability to do this stringification in certain circumstances such as that of emitting DDL. In order to access this functionality one can use the 'literal_binds' flag, passed to compile_kwargs:
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}))
上面的方法需要注意的是,它只支持基本的 类型,例如int和字符串,以及bindparam 如果不直接使用预先设置的值,则无法实现 也可以Stringify它。
若要支持不支持的类型的内联文字呈现,请实现 类的目标类型的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)
为此,我们可以使用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语句 数据库。
这段代码是基于@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)
根据文档。
只是一个简单的彩色示例与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()))