如何从 SQLAlchemy 表达式中获取原始的、已编译的 SQL 查询？-IT科技

摘要：问题描述：我有一个 SQLAlchemy 查询对象，想要获取已编译 SQL 语句的文本，以及绑定的所有参数（例如，没有%s或其他变量等待由语句编译器或 MySQLdb 方言引擎等绑定）。调用查询str()会显示如下内容：SELECT id WHERE date_added <= %s AND date_...

问题描述：

我有一个 SQLAlchemy 查询对象，想要获取已编译 SQL 语句的文本，以及绑定的所有参数（例如，没有%s或其他变量等待由语句编译器或 MySQLdb 方言引擎等绑定）。

调用查询str()会显示如下内容：

SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC

我尝试过在 query._params 中查找，但它是一个空的字典。我使用装饰器的这个示例sqlalchemy.ext.compiler.compiles编写了自己的编译器，但那里的语句仍然包含%s我想要的数据。

我不太清楚何时将参数混合在一起以创建查询；检查查询对象时，它们始终是一个空字典（尽管查询执行正常，并且打开回显日志记录时引擎会将其打印出来）。

我开始意识到 SQLAlchemy 不想让我知道底层查询，因为它破坏了表达式 API 接口与所有不同 DB-API 的通用性质。我不介意在我发现查询是什么之前执行查询；我只是想知道！

解决方案 1：

该文档用于literal_binds打印包含参数的查询q：

print(q.statement.compile(compile_kwargs={"literal_binds": True}))

上述方法的缺点是它只支持基本类型，比如整数和字符串，而且如果直接使用没有预设值的 bindparam()，它也无法将其字符串化。

该文档还发出以下警告：

切勿将此技术用于从不受信任的输入（例如来自 Web 表单或其他用户输入应用程序）接收的字符串内容。SQLAlchemy 将 Python 值强制转换为直接 SQL 字符串值的功能对于不受信任的输入并不安全，并且不会验证传递的数据类型。以编程方式调用针对关系数据库的非 DDL SQL 语句时，请始终使用绑定参数。

解决方案 2：

Nicolas Cadou 的这篇博文提供了最新答案。

引用博客文章，这是对我的建议和有用的：

from sqlalchemy.dialects import postgresql
print str(q.statement.compile(dialect=postgresql.dialect()))

其中 q 定义为：

q = DBSession.query(model.Name).distinct(model.Name.value) \n>              .order_by(model.Name.value)

或者只是任何一种session.query()。

解决方案 3：

这应该适用于 Sqlalchemy >= 0.6

from sqlalchemy.sql import compiler

from psycopg2.extensions import adapt as sqlescape
# or use the appropiate escape function from your db driver

def compile_query(query):
    dialect = query.session.bind.dialect
    statement = query.statement
    comp = compiler.SQLCompiler(dialect, statement)
    comp.compile()
    enc = dialect.encoding
    params = {}
    for k,v in comp.params.iteritems():
        if isinstance(v, unicode):
            v = v.encode(enc)
        params[k] = sqlescape(v)
    return (comp.string.encode(enc) % params).decode(enc)

解决方案 4：

问题是，sqlalchemy 永远不会将数据与查询混合。查询和数据分别传递给底层数据库驱动程序 - 数据的插值发生在数据库中。

Sqlalchemy 将您所见的查询传递到str(myquery)数据库中，并且值将进入单独的元组。

您可以使用某种方法自己使用查询插入数据（正如 albertov 在下面建议的那样），但这与 sqlalchemy 执行的操作不同。

解决方案 5：

对于 MySQLdb 后端，我修改了 albertov 的精彩答案（非常感谢！）。我确信它们可以合并以检查是否comp.positional如此，True但这略微超出了这个问题的范围。

def compile_query(query):
    from sqlalchemy.sql import compiler
    from MySQLdb.converters import conversions, escape

    dialect = query.session.bind.dialect
    statement = query.statement
    comp = compiler.SQLCompiler(dialect, statement)
    comp.compile()
    enc = dialect.encoding
    params = []
    for k in comp.positiontup:
        v = comp.params[k]
        if isinstance(v, unicode):
            v = v.encode(enc)
        params.append( escape(v, conversions) )
    return (comp.string.encode(enc) % tuple(params)).decode(enc)

解决方案 6：

首先，我先声明一下，我假设您这样做主要是为了调试目的——我不建议您尝试在 SQLAlchemy 流畅 API 之外修改语句。

不幸的是，似乎没有一种简单的方法来显示包含查询参数的编译语句。 SQLAlchemy 实际上并没有将参数放入语句中——它们作为字典传递到数据库引擎中。这让特定于数据库的库可以处理诸如转义特殊字符以避免 SQL 注入之类的事情。

但您可以相当轻松地通过两步过程完成此操作。要获取语句，您可以按照已经显示的方式进行操作，然后打印查询：

>>> print(query)
SELECT field_1, field_2 FROM table WHERE id=%s;

您可以使用 query.statement 更进一步查看参数名称。注意:id_1下面与%s上面的区别——在这个非常简单的例子中，这不是什么问题，但在更复杂的语句中可能是关键。

>>> print(query.statement)
>>> print(query.statement.compile()) # seems to be equivalent, you can also
                                     # pass in a dialect if you want
SELECT field_1, field_2 FROM table WHERE id=:id_1;

params然后，您可以通过获取已编译语句的属性来获取参数的实际值：

>>> print(query.statement.compile().params)
{u'id_1': 1}

这至少对 MySQL 后端有效；我希望它对 PostgreSQL 也足够通用，而不需要使用psycopg2。

解决方案 7：

对于使用 psycopg2 的 postgresql 后端，您可以监听事件do_execute，然后使用游标、语句和类型强制参数以及Cursor.mogrify()内联参数。您可以返回 True 以阻止实际执行查询。

import sqlalchemy

class QueryDebugger(object):
    def __init__(self, engine, query):
        with engine.connect() as connection:
            try:
                sqlalchemy.event.listen(engine, "do_execute", self.receive_do_execute)
                connection.execute(query)
            finally:
                sqlalchemy.event.remove(engine, "do_execute", self.receive_do_execute)

    def receive_do_execute(self, cursor, statement, parameters, context):
        self.statement = statement
        self.parameters = parameters
        self.query = cursor.mogrify(statement, parameters)
        # Don't actually execute
        return True

使用示例：

>>> engine = sqlalchemy.create_engine("postgresql://postgres@localhost/test")
>>> metadata = sqlalchemy.MetaData()
>>> users = sqlalchemy.Table('users', metadata, sqlalchemy.Column("_id", sqlalchemy.String, primary_key=True), sqlalchemy.Column("document", sqlalchemy.dialects.postgresql.JSONB))
>>> s = sqlalchemy.select([users.c.document.label("foobar")]).where(users.c.document.contains({"profile": {"iid": "something"}}))
>>> q = QueryDebugger(engine, s)
>>> q.query
'SELECT users.document AS foobar 
FROM users 
WHERE users.document @> \'{"profile": {"iid": "something"}}\''
>>> q.statement
'SELECT users.document AS foobar 
FROM users 
WHERE users.document @> %(document_1)s'
>>> q.parameters
{'document_1': '{"profile": {"iid": "something"}}'}

解决方案 8：

以下解决方案使用 SQLAlchemy 表达式语言，适用于 SQLAlchemy 1.1。此解决方案不会将参数与查询混合（如原作者所要求），但提供了一种使用 SQLAlchemy 模型为不同的 SQL 方言生成 SQL 查询字符串和参数字典的方法。该示例基于教程http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

鉴于班级，

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class foo(Base):
    __tablename__ = 'foo'
    id = Column(Integer(), primary_key=True)
    name = Column(String(80), unique=True)
    value = Column(Integer())

我们可以使用select函数生成查询语句。

from sqlalchemy.sql import select    
statement = select([foo.name, foo.value]).where(foo.value > 0)

接下来，我们可以将该语句编译成查询对象。

query = statement.compile()

默认情况下，该语句使用与 SQLite 和 Oracle 等 SQL 数据库兼容的基本“命名”实现进行编译。如果您需要指定方言（例如 PostgreSQL），则可以执行

from sqlalchemy.dialects import postgresql
query = statement.compile(dialect=postgresql.dialect())

或者如果您想明确指定方言为 SQLite，您可以将 paramstyle 从“qmark”更改为“named”。

from sqlalchemy.dialects import sqlite
query = statement.compile(dialect=sqlite.dialect(paramstyle="named"))

从查询对象中，我们可以提取查询字符串和查询参数

query_str = str(query)
query_params = query.params

最后执行查询。

conn.execute( query_str, query_params )

解决方案 9：

更新：又出现了另一种情况，即先前的解决方案无法正确生成正确的 SQL 语句。在 SQLAlchemy 中深入研究后，很明显您不仅需要针对特定方言进行编译，还需要获取已编译的查询并将其初始化为正确的 DBAPI 连接上下文。否则，类型绑定处理器之类的东西就不会执行，JSON.NULL 之类的值也不会得到正确转换。

请注意，这使得此解决方案非常特定于 Flask + Flask-SQLAlchemy + psycopg2 + PostgreSQL。您可能需要通过更改方言和引用连接的方式将此解决方案翻译到您的环境中。但是，我非常有信心这会为所有数据类型生成精确的 SQL。

下面的结果是一种简单的方法，只需查询查询本身，就可以偶尔但可靠地获取精确的、编译好的 SQL，并将其发送到我的 PostgreSQL 后端：

import sqlalchemy.dialects.postgresql.psycopg2

from flask import current_app

def query_to_string(query):
    dialect = sqlalchemy.dialects.postgresql.psycopg2.dialect()
    compiled_query = query.statement.compile(dialect=dialect)
    sqlalchemy_connection = current_app.db.session.connection()
    context = dialect.execution_ctx_cls._init_compiled(
        dialect,
        sqlalchemy_connection,
        sqlalchemy_connection.connection,
        compiled_query,
        None
    )
    mogrified_query = sqlalchemy_connection.connection.cursor().mogrify(
        context.statement,
        context.parameters[0]
    )
    return mogrified_query.decode()

query = [ .... some ORM query .... ]

print(f"compiled SQL = {query_to_string(query)}")

解决方案 10：

您可以使用ConnectionEvents系列中的事件：after_cursor_execute或before_cursor_execute。

在@zzzeek 的sqlalchemy UsageRecipes中你可以找到这个示例：

Profiling

...
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement,
                        parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())
    logger.debug("Start Query: %s" % statement % parameters)
...

您可以在这里查看您的对账单

解决方案 11：

我已经创建了这个小函数，当我想打印完整查询时我会导入它，考虑到我正处于测试中，此时方言已经被绑定：

import re

def print_query(query):
    regex = re.compile(":(?P<name>w+)")
    params = query.statement.compile().params
    sql = regex.sub("'{g<name>}'", str(query.statement)).format(**params)
    print(f"
Printing SQLAlchemy query:

")
    print(sql)
    return sql

解决方案 12：

我认为 .statement 可能会起作用：
http://docs.sqlalchemy.org/en/latest/orm/query.html ?highlight=query

>>> local_session.query(sqlalchemy_declarative.SomeTable.text).statement
<sqlalchemy.sql.annotation.AnnotatedSelect at 0x6c75a20; AnnotatedSelectobject>
>>> x=local_session.query(sqlalchemy_declarative.SomeTable.text).statement
>>> print(x)
SELECT sometable.text 
FROM sometable

解决方案 13：

如果您使用 SQLAlchemy 来使用 PyMySQL，那么您可以使用一个技巧。

由于我太着急，浪费了很多时间，所以我把驱动程序改成了用参数打印当前语句。

SQLAlchemy 有意不支持文字值的完全字符串化。

但是 PyMySQL 有 'mogrify' 方法可以做到这一点，但是，SQLALchemy 没有 HOOK 可以在使用 ORM 插入/更新（当它控制游标时）时调用它，例如 db.add 或 commit/flush（用于更新）。

因此，只需前往驱动程序正在使用的地方（即可知道在哪里使用）：pip show pycharm

在文件夹中，找到并编辑文件cursors.py。

在方法中：

def execute(self, query, args=None):

底下是：

query = self.mogrify(query, args)

只需添加：

print(query)

将像魔法一样工作，调试，解决问题并删除打印。