SQLAlchemy：打印实际查询

Question

我真的希望能打印出我应用程序的有效SQL语句，包括具体的值，而不是绑定参数。不过，在SQLAlchemy中怎么做到这一点并不明显（我觉得这也是故意设计的）。

有没有人以一种通用的方式解决了这个问题？

Answer 1

如果你想在调试的时候查看 SQL 查询的情况，可以在启动 SQLAlchemy 时加上 echo=True，这样就能记录下所有的 SQL 查询。例如：

engine = create_engine(
    "mysql://scott:tiger@hostname/dbname",
    encoding="latin1",
    echo=True,
)

你也可以只对某一次请求进行这样的设置：

echo=False – 如果设置为 True，那么引擎会把所有的 SQL 语句和它们的参数列表记录到日志中，默认是输出到 sys.stdout。你可以随时修改 Engine 的 echo 属性来开启或关闭日志记录。如果把它设置为字符串 "debug"，那么结果行也会打印到标准输出。这项设置实际上控制了一个 Python 的日志记录器；想了解如何直接配置日志记录，可以查看 配置日志记录。

来源：SQLAlchemy 引擎配置

如果你在使用 Flask，可以简单地设置

app.config["SQLALCHEMY_ECHO"] = True

这样就能得到相同的效果。

Answer 2

在绝大多数情况下，将 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 适当地处理。此外，绕过绑定参数可能是现代网络应用中最常见的安全漏洞。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}))

上述方法有一些限制，只支持基本类型，比如整数和字符串。此外，如果直接使用一个没有预设值的 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)

Answer 3

这个在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