我确实希望能够为我的应用程序打印有效的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语句 数据库。

在绝大多数情况下,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)

我想指出,上面给出的解决方案并不“只适用于”非平凡的查询。我遇到的一个问题是更复杂的类型,比如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)

根据文档。

只是一个简单的彩色示例与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()))