如何从SQLAlchemy表达式获取原始的编译SQL查询?

问题:如何从SQLAlchemy表达式获取原始的编译SQL查询?

我有一个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我想要的数据。

我无法弄清楚何时混入参数来创建查询。在检查查询对象时,它们始终是一个空字典(尽管查询执行得很好,并且当您打开echo记录时引擎会打印出来)。

我开始收到消息,SQLAlchemy不想让我知道底层查询,因为它破坏了表达式API接口的所有不同DB-API的一般性质。我不在乎查询是否在我发现查询之前就已经执行了;我只是想知道!

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 %s or other variables waiting to be bound by the statement compiler or MySQLdb dialect engine, etc).

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

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.compiles decorator but even the statement there still has %s where I want data.

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!


回答 0

博客提供了更新的答案。

引用博客文章中的内容,这对我来说是建议和有效的。

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

其中q定义为:

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

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

感谢Nicolas Cadou的回答!希望对其他在这里搜索的人有所帮助。

This blog 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 = DBSession.query(model.Name).distinct(model.Name.value) \
             .order_by(model.Name.value)

Or just any kind of session.query().

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


回答 1

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

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

上面的方法有一个警告:仅基本类型(例如int和字符串)才支持该方法,此外,如果直接使用没有预设值的bindparam(),则也不能将其字符串化。

该文档还发出以下警告:

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

The documentation uses literal_binds to print a query q including parameters:

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.

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.


回答 2

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

This should work with 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)

回答 3

对于MySQLdb后端,我稍微修改了albertov的出色答案(非常感谢!)。我敢肯定,可以将它们合并以检查是否存在comp.positionalTrue但这超出了此问题的范围。

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)

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.positional was True but that’s slightly beyond the scope of this question.

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)

回答 4

事实是,sqlalchemy永远不会将数据与查询混合在一起。查询和数据分别传递到基础数据库驱动程序-数据插值发生在数据库中。

Sqlalchemy如您所见将查询传递str(myquery)给数据库,并且值将进入一个单独的元组。

您可以使用一些方法自己在查询中插入数据(如下面的albertov所建议),但这与sqlalchemy正在执行的事情不同。

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 passes the query as you’ve seen in str(myquery) to the database, and the values will go in a separate tuple.

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.


回答 5

首先,让我先说一下,我假设您这样做主要是出于调试目的-我不建议您尝试尝试在SQLAlchemy Fluent 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

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.

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.

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_1 below vs %s above — not really a problem in this very simple example, but could be key in a more complicated statement.

>>> 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;

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

>>> 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.


回答 6

对于使用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 \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"}}'}

For postgresql backend using psycopg2, you can listen for the do_execute event, 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.

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"}}'}

回答 7

以下解决方案使用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,则可以将参数样式从“ 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 )

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

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 select function.

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

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’.

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 )

回答 8

您可以使用ConnectionEvents系列的事件:after_cursor_executebefore_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)
...

在这里您可以访问您的对帐单

You can use events from ConnectionEvents family: after_cursor_execute or before_cursor_execute.

In sqlalchemy UsageRecipes by @zzzeek you can find this example:

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


回答 9

因此,将这些不同答案的很多点放在一起,我得出了我需要的东西:一组简单的代码可以插入,偶尔但可靠地(即处理所有数据类型)获取发送给我的准确的,已编译的SQL通过查询查询本身的Postgres后端:

from sqlalchemy.dialects import postgresql

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

compiled_query = query.statement.compile(
    dialect=postgresql.dialect()
)
mogrified_query = session.connection().connection.cursor().mogrify(
    str(compiled_query),
    compiled_query.params
)

print("compiled SQL = {s}".format(mogrified_query.decode())

So, putting together a lot of little bits of these different answers, I came up with what I needed: a simple set of code to drop in and occasionally but reliably (i.e. handles all data types) grab the exact, compiled SQL sent to my Postgres backend by just interrogating the query itself:

from sqlalchemy.dialects import postgresql

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

compiled_query = query.statement.compile(
    dialect=postgresql.dialect(),
    compile_kwargs={"literal_binds": True}
)
mogrified_query = session.connection().connection.cursor().mogrify(
    str(compiled_query),
    compiled_query.params
)

print("compiled SQL = {s}".format(mogrified_query.decode())

回答 10

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

I think .statement would possibly do the trick: 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