Python 如何从 SQLAlchemy 表达式中获取原始的、编译的 SQL 查询？

How do I get a raw, compiled SQL query from a SQLAlchemy expression?

提问by cce

I have a SQLAlchemy query object and want to get the text of the compiled SQL statement, with all its parameters bound (e.g. no %sor other variables waiting to be bound by the statement compiler or MySQLdb dialect engine, etc).

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

Calling str()on the query reveals something like this:

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

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

I've tried looking in query._params but it's an empty dict. I wrote my own compiler using this example of the sqlalchemy.ext.compiler.compilesdecoratorbut even the statement there still has %swhere I want data.

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

I can't quite figure out when my parameters get mixed in to create the query; when examining the query object they're always an empty dictionary (though the query executes fine and the engine prints it out when you turn echo logging on).

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

I'm starting to get the message that SQLAlchemy doesn't want me to know the underlying query, as it breaks the general nature of the expression API's interface all the different DB-APIs. I don't mind if the query gets executed before I found out what it was; I just want to know!

我开始收到消息，SQLAlchemy 不希望我知道底层查询，因为它破坏了所有不同 DB-API 的表达式 API 接口的一般性质。我不介意在我发现它是什么之前查询是否被执行；我只是想知道！

采纳答案by AndyBarr

Thisblog provides an updated answer.

该博客提供了更新的答案。

Quoting from the blog post, this is suggested and worked for me.

引用博客文章，这是建议并为我工作。

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

Where q is defined as:

其中 q 定义为：

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

Or just any kind of session.query().

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

Thanks to Nicolas Cadou for the answer! I hope it helps others who come searching here.

感谢 Nicolas Cadou 的回答！我希望它可以帮助到这里搜索的其他人。

回答by albertov

This should work with Sqlalchemy >= 0.6

这应该适用于 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)

回答by nosklo

Thing is, sqlalchemy never mixes the data with your query. The query and the data are passed separately to your underlying database driver - the interpolation of data happens in your database.

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

Sqlalchemy passes the query as you've seen in str(myquery)to the database, and the values will go in a separate tuple.

Sqlalchemy 将您str(myquery)在数据库中看到的查询传递给数据库，并且这些值将放在一个单独的元组中。

You could use some approach where you interpolate the data with the query yourself (as albertov suggested below), but that's not the same thing that sqlalchemy is executing.

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

回答by cce

For the MySQLdb backend I modified albertov's awesome answer (thanks so much!) a bit. I'm sure they could be merged to check if comp.positionalwas Truebut that's slightly beyond the scope of this question.

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

回答by user2757902

I think .statement would possibly do the trick: http://docs.sqlalchemy.org/en/latest/orm/query.html?highlight=query

我认为 .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

回答by Alex Bender

You can use events from ConnectionEventsfamily: after_cursor_executeor before_cursor_execute.

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

In sqlalchemy UsageRecipesby @zzzeek you can find this example:

在@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)
...

Here you can get access to your statement

在这里您可以访问您的声明

回答by Matt

The documentationuses literal_bindsto print a query qincluding parameters:

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

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

the above approach has the caveats that it is only supported for basic types, such as ints and strings, and furthermore if a bindparam() without a pre-set value is used directly, it won't be able to stringify that either.

上面的方法有一个警告，它只支持基本类型，例如整数和字符串，而且如果直接使用没有预设值的 bindparam() ，它也无法将其字符串化。

The documentation also issues this warning:

该文档还发出此警告：

Never use this technique with string content received from untrusted input, such as from web forms or other user-input applications. SQLAlchemy's facilities to coerce Python values into direct SQL string values are not secure against untrusted input and do not validate the type of data being passed. Always use bound parameters when programmatically invoking non-DDL SQL statements against a relational database.

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

回答by eric

The following solution uses the SQLAlchemy Expression Language and works with SQLAlchemy 1.1. This solution does not mix the parameters with the query (as requested by the original author), but provides a way of using SQLAlchemy models to generate SQL query strings and parameter dictionaries for different SQL dialects. The example is based on the tutorial http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

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

Given the class,

鉴于班级，

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

we can produce a query statement using the selectfunction.

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

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

Next, we can compile the statement into a query object.

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

query = statement.compile()

By default, the statement is compiled using a basic 'named' implementation that is compatible with SQL databases such as SQLite and Oracle. If you need to specify a dialect such as PostgreSQL, you can do

默认情况下，该语句是使用与 SQLite 和 Oracle 等 SQL 数据库兼容的基本“命名”实现编译的。如果需要指定像PostgreSQL这样的方言，可以这样做

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

Or if you want to explicitly specify the dialect as SQLite, you can change the paramstyle from 'qmark' to 'named'.

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

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

From the query object, we can extract the query string and query parameters

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

query_str = str(query)
query_params = query.params

and finally execute the query.

最后执行查询。

conn.execute( query_str, query_params )

回答by rectalogic

For postgresql backend using psycopg2, you can listen for the do_executeevent, then use the cursor, statement and type coerced parameters along with Cursor.mogrify()to inline the parameters. You can return True to prevent actual execution of the query.

对于使用 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

Sample usage:

示例用法：

>>> 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 \nFROM users \nWHERE users.document @> \'{"profile": {"iid": "something"}}\''
>>> q.statement
'SELECT users.document AS foobar \nFROM users \nWHERE users.document @> %(document_1)s'
>>> q.parameters
{'document_1': '{"profile": {"iid": "something"}}'}

回答by Hannele

First let me preface by saying that I assume you're doing this mainly for debugging purposes -- I wouldn't recommend trying to modify the statement outside of the SQLAlchemy fluent API.

首先让我先说我假设您这样做主要是为了调试目的——我不建议尝试修改 SQLAlchemy fluent API 之外的语句。

Unfortunately there doesn't seem to be a simple way to show the compiled statement with the query parameters included. SQLAlchemy doesn't actually put the parameters into the statement -- they're passed into the database engine as a dictionary. This lets the database-specific library handle things like escaping special characters to avoid SQL injection.

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

But you can do this in a two-step process reasonably easily. To get the statement, you can do as you've already shown, and just print the query:

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

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

You can get one step closer with query.statement, to see the parameter names. (Note :id_1below vs %sabove -- not really a problem in this very simple example, but could be key in a more complicated statement.)

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

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

Then, you can get the actual values of the parameters by getting the paramsproperty of the compiled statement:

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

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

This worked for a MySQL backend at least; I would expect it's also general enough for PostgreSQL without needing to use psycopg2.

这至少适用于 MySQL 后端；我希望它对于 PostgreSQL 也足够通用，而无需使用psycopg2.