SQLAlchemy:打印实际的查询
我真的希望能够为我的应用程序打印出有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(按照devise,我相当确定)。
有没有人以一般的方式解决这个问题?
在绝大多数情况下,SQLAlchemy语句或查询的“string化”如下所示:
print str(statement)
这适用于ORM Query
以及任何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)
关于绑定参数要被“内联”到最终string的原始规定,这里的挑战是SQLAlchemy通常不会被分配任务,因为这是由Python DBAPI适当地处理的,更不用说绕过绑定参数了可能是现代Web应用程序中使用最广泛的安全漏洞。 SQLAlchemy在某些情况下(例如发出DDL)的能力有限。 为了访问这个function,可以使用'literal_binds'标志,传递给compile_kwargs
:
from sqlalchemy.sql import table, column, select t = table('t', column('x')) s = select([t]).where(tcx == 5) print s.compile(compile_kwargs={"literal_binds": True})
上面的方法有一些注意事项,它只支持基本types,例如整数和string,而且如果没有预设值的bindparam
被直接使用,它将不能够对其进行string化。
要支持对不支持的types进行内联文字渲染, TypeDecorator
为包含TypeDecorator.process_literal_param
方法的目标types实现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.cx > 5).compile( compile_kwargs={"literal_binds": True}) )
产量如下:
SELECT mytable.x FROM mytable WHERE mytable.x > my_fancy_formatting(5)
这在Python 2和3中工作,比以前有点干净,但要求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中testing)
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辉煌的现有答案 。 我只是将datetime.datetime
types的自定义渲染添加到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)
所以build立@ 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
重新sqlparse
了SQL。 它可以使用pip install sqlparse
进行安装。
我想指出的是,上面给出的解决scheme并不是“非常有用”的,而且是不重要的查询。 我遇到的一个问题是更复杂的types,例如pgsql ARRAY导致问题。 我确实find了一个解决scheme,对我来说,即使使用pgsql ARRAY也行:
借用: https : //gist.github.com/gsakkis/4572159
链接的代码似乎基于SQLAlchemy的旧版本。 你会得到一个错误说,属性_mapper_zero_or_none不存在。 这是一个更新的版本,将与更新的版本一起使用,您只需使用bindreplace_mapper_zero_or_none即可。 另外,它支持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)
testing到两个层次的嵌套数组。
我们可以使用编译方法来达到这个目的。 从文档 :
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'