SQLAlchemy：打印实际查询

Question

SQLAlchemy：打印实际查询

pythonsqlalchemy

294

我希望能够打印出包括值而非绑定参数在内的有效SQL，但在SQLAlchemy中如何实现并不明显（我相信这是有意为之设计的）。

是否有人已经以通用的方式解决了这个问题？

- bukzor

1

我没有尝试过，但你可能可以通过利用SQLAlchemy的sqlalchemy.engine日志来构建一个更加稳健的解决方案。它记录查询和绑定参数，你只需要将绑定占位符替换为已准备好的SQL查询字符串上的值即可。 - Simon

@Simon：使用记录器存在两个问题：1）它仅在语句执行时打印；2）我仍然需要进行字符串替换，但在这种情况下，我不会完全知道绑定模板字符串，我必须从查询文本中解析出来，使解决方案更加脆弱。 - bukzor

新的URL似乎是http://docs.sqlalchemy.org/en/latest/faq/sqlexpressions.html#how-do-i-render-sql-expressions-as-strings-possibly-with-bound-parameters-inlined，用于@zzzeek的FAQ。 - Jim DeLaHunt

11个回答

144

在调试时，只有当你需要时才有意义，你可以使用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配置

如果在Flask中使用，则只需设置

app.config["SQLALCHEMY_ECHO"] = True

达到相同的行为。

- Vedran Šego

18

这个回答应该要得到更高的评价。对于flask-sqlalchemy用户来说，这应该是被采纳的答案。 - jso

是的，这是“正确”的答案。也许它在2014年之前不存在？ - Mike Williamson

@MikeWilliamson，你可以在这里找到它的使用方法（https://docs.sqlalchemy.org/en/13/changelog/migration_08.html），这是一个2012年10月的文档，更新于2013年3月。不可否认，这比问题发布时间（2011年4月）要新一些。 - Vedran Šego

78

这适用于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中测试通过）

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

7

太棒了！我们需要将这个加入一些调试库，以便我们可以轻松访问它。感谢您在这件事上的努力。我很惊讶它竟然如此复杂。 - Corey O.

8

我很确定这是故意设置为难以理解的，因为新手容易使用 cursor.execute() 这个字符串。但原则上Python中通常使用“成年人自愿原则”。 - bukzor

非常好。我冒昧地将其合并到https://dev59.com/IF0Z5IYBdhLWcg3w3DUW#42066590中，该页面涵盖了SQLAlchemy v0.7.9-v1.1.15，包括INSERT和UPDATE语句（PY2 / PY3）。 - wolfmanx

非常好。但是它是否转换如下。 1）query（Table）。filter（Table.Column1.is_（False）到WHERE Column1 IS 0。 2）query（Table）。filter（Table.Column1.is_（True）到WHERE Column1 IS 1。 3）query（Table）。filter（Table.Column1 == func.any（[1,2,3]）到WHERE Column1 = any（'[1,2,3]'）。以上转换在语法上是不正确的。 - Sekhar C

如何解决插入语句中的默认值？

from sqlalchemy import MetaData, insert
from sqlalchemy.schema import Column, Table

meta = MetaData()

TGR = Table('tbl', meta,
        Column('a'),
        Column('b'),
        Column('c', default='xxx')
)

s = TGR.insert().values(a=1, b=2)
print(literalquery(s))

INSERT INTO tbl (a, b, c) VALUES (1, 2, :c) - janezj

这绝对是我最喜欢的解决方案，但最近我一直遇到以下错误： init()收到了一个意外的关键字参数'_enums' 有任何想法为什么它无法编译查询吗？ - c8999c 3f964f64

57

我们可以使用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'

来自文档的警告：

永远不要将来自不可信来源（例如来自网络表单或其他用户输入应用程序）的字符串内容与此技术一起使用。SQLAlchemy的将Python值强制转换为直接SQL字符串值的功能不安全，无法验证传递的数据类型。在针对关系数据库编程调用非DDL SQL语句时，始终使用绑定参数。

- akshaynagpal

请参考此页面获取方言信息：https://docs.sqlalchemy.org/en/13/dialects/ - Taras

1

这是最好的答案。 - Ping.Goblue

1

你可能会遇到 *** NotImplementedError: 不知道如何对值 '2022-09-29 18:50:00 +0200' 进行文字引用。 - tread

1

@tread，我正在结合这个答案和下面bukzor的答案。由于两种解决方案在某些情况下会产生错误，因此我只是连续使用两种方言进行尝试-除外处理，首先是mysql/postgres方言，然后是LiteralDialect（见下文）。 - c8999c 3f964f64

19

在 @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 安装。

- jmagnusson

@bukzor 所有值都可以正常工作，除了在使用 Python 3 + SQLAlchemy 1.0 时的 datatime.now()。你需要遵循 @zzzeek 的建议，为此创建一个自定义 TypeDecorator 才能使其正常工作。 - jmagnusson

这有点太具体了。日期时间在任何Python和SQLAlchemy的组合中都不起作用。此外，在py27中，非ASCII Unicode会导致爆炸。 - bukzor

据我所见，TypeDecorator路线要求我更改表定义，这并不是仅仅查看我的查询的合理要求。我编辑了我的答案，使其与您和zzzeek的答案更接近，但我采取了自定义方言的路线，这与表定义完全无关。 - bukzor

14

这段代码基于@bukzor的优秀答案。我只是为Oracle的TO_DATE()添加了datetime.datetime类型的自定义渲染。

请随意更新代码以适应您的数据库：

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)

- vvladymyrov

45

我不明白南非人为什么认为这样一个简单的操作会如此困难是合理的。 - bukzor

这个配方（以及原始配方）会在任何bindparam字符串值无法表示为ascii时引发UnicodeDecodeError。我发布了一个gist来解决这个问题。 - gsakkis

1

"STR_TO_DATE('%s','%%Y-%%m-%%d %%H:%%M:%%S')" % value.strftime("%Y-%m-%d %H:%M:%S") 在mysql中的含义是将时间字符串转换为日期格式。 - Zitrax

1

@bukzor - 我不记得有人问过我上面的内容是否“合理”，所以你不能真正地说我“相信”它是 - 顺便说一句，它并不是！ :) 请看我的回答。 - zzzeek

你怎么知道Python DBAPI驱动程序用于绑定datetime.datetime类型的参数是什么？你为什么认为它是TO_DATE()？ - Piotr Dobrogost

显示剩余2条评论

11

我想指出，以上给出的解决方案并不能在复杂查询中“简单运行”。我遇到的一个问题是更复杂的类型，比如pgsql ARRAYs会引起问题。我找到了一个解决方案，对我而言，即使有pgsql ARRAYs也可以正常工作：

参考： https://gist.github.com/gsakkis/4572159

链接的代码似乎基于较旧版本的SQLAlchemy。您将收到一个错误，指出属性_mapper_zero_or_none不存在。这里是一个更新后的版本，可以与新版本一起使用，只需用bind替换_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 https://dev59.com/8m035IYBdhLWcg3wGL5T
    """
    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)

测试了两层嵌套数组。

- JamesHutchison

请展示一个如何使用它的例子？谢谢。 - slashdottir

from file import render_query; print(render_query(query)) - Alfonso Embid-Desmet

这是整个页面中唯一一个对我有效的示例！谢谢！ - fougerejo

long 是从哪里导入的？ - Talha Junaid

使用 int 代替 long。在 Python 3 中，long 已被弃用。 - Samkit Jain

6

To log SQL queries using Python logging instead of the echo=True flag:
import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

根据文档。

- EliadL

你好。我已经阅读了文档，但是不太清楚在哪里可以启用我们引擎的日志记录功能... - undefined

3

一个使用ORM查询和pygments的简单着色示例。

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()))

- Mark Mishyn

0

由于某些原因，print(str(stmt)) 对我无效。

(也许是因为我在运行测试时尝试打印？我发现调试打印功能有点受影响。)

然而有趣的是，这个有效：

stmt.__str__()

- half of a glazier

嗨。抱歉，但是stmt具体代表什么？是用于查询的对象吗？例如：product.query？ - undefined

1

@RaulChiarella 参考@akshaynagpal的回答，这是一个语句的示例。 - undefined

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，

- zzzeek · Accepted Answer

在绝大多数情况下，将SQLAlchemy语句或查询转换为字符串非常简单：

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))

关于原始规定，绑定参数需要“内联”到最终字符串中，这里的挑战是SQLAlchemy通常不需要处理此操作，因为这已由Python DBAPI正确处理，更不用说绕过绑定参数可能是现代Web应用程序中最广泛利用的安全漏洞之一。SQLAlchemy在某些情况下具有限制能力，例如发出DDL时。为了访问此功能，可以使用传递给compile_kwargs的“literal_binds”标志：

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和string。此外，如果直接使用未预设值的bindparam，也无法将其字符串化。为了支持不受支持的类型的内联文字呈现，需要为目标类型实现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)