使用Python日志代替echo=True标记记录SQL查询: 导入日志 logging.basicConfig () logging.getLogger (sqlalchemy.engine) .setLevel (logging.INFO)

根据文档。

为此，我们可以使用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语句 数据库。

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

所以基于@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安装它。

使用Python日志代替echo=True标记记录SQL查询: 导入日志 logging.basicConfig () logging.getLogger (sqlalchemy.engine) .setLevel (logging.INFO)

根据文档。

这可以在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