我确实希望能够为我的应用程序打印有效的SQL,包括值,而不是绑定参数,但是在SQLAlchemy中如何做到这一点并不明显(我相当肯定是设计上的)。

有人用一般的方法解决了这个问题吗?


当前回答

这就是我的方法

# query is instance of: from sqlalchemy import select
def raw_query(query):
    q = str(query.compile())
    p = query.compile().params
    for k in p.keys():
        v = p.get(k)
        if isinstance(v, (int, float, complex)):
            q = q.replace(f":{k}", f"{v}")
        else:
            q = q.replace(f":{k}", f"'{v}'")
    print(q)

如何使用:

from sqlalchemy import select

select_query = select([
    any_model_table.c["id_account"],
    any_model_table.c["id_provider"],
    any_model_table.c["id_service"],
    func.sum(any_model_table.c["items"]).label("items"),
    # @eaf
    func.date_format(func.now(), "%Y-%m-%d").label("some_date"),
    func.date_format(func.now(), "%Y").label("as_year"),
    func.date_format(func.now(), "%m").label("as_month"),
    func.date_format(func.now(), "%d").label("as_day"),
]).group_by(
    any_model_table.c.id_account,
    any_model_table.c.id_provider,
    any_model_table.c.id_service
).where(
    any_model_table.c.id == 5

).where(
    func.date_format(any_model_table.c.dt, "%Y-%m-%d") == datetime.utcnow().strftime('%Y-%m-%d')
)

raw_query(select_query)

其他回答

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

这就是我的方法

# query is instance of: from sqlalchemy import select
def raw_query(query):
    q = str(query.compile())
    p = query.compile().params
    for k in p.keys():
        v = p.get(k)
        if isinstance(v, (int, float, complex)):
            q = q.replace(f":{k}", f"{v}")
        else:
            q = q.replace(f":{k}", f"'{v}'")
    print(q)

如何使用:

from sqlalchemy import select

select_query = select([
    any_model_table.c["id_account"],
    any_model_table.c["id_provider"],
    any_model_table.c["id_service"],
    func.sum(any_model_table.c["items"]).label("items"),
    # @eaf
    func.date_format(func.now(), "%Y-%m-%d").label("some_date"),
    func.date_format(func.now(), "%Y").label("as_year"),
    func.date_format(func.now(), "%m").label("as_month"),
    func.date_format(func.now(), "%d").label("as_day"),
]).group_by(
    any_model_table.c.id_account,
    any_model_table.c.id_provider,
    any_model_table.c.id_service
).where(
    any_model_table.c.id == 5

).where(
    func.date_format(any_model_table.c.dt, "%Y-%m-%d") == datetime.utcnow().strftime('%Y-%m-%d')
)

raw_query(select_query)

这段代码是基于@bukzor现有的精彩答案。我刚刚为datetime添加了自定义渲染。datetime类型到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)

假定只有在调试时才有意义,可以使用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 Configuration

如果与Flask一起使用,您可以简单地设置

app.config["SQLALCHEMY_ECHO"] = True

得到相同的行为。

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

根据文档。