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

Question

我有一个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的一般性质.我不介意在我发现它之前是否执行了查询; 我只是想知道!

Answer 1

此博客提供了更新的答案.

从博客文章引用,这是建议并为我工作.

>>> 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的回答!我希望它可以帮助那些来这里搜索的人.

Answer 2

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

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

上面的方法有一些警告,它只支持基本类型,例如int和字符串,而且如果直接使用没有预设值的bindparam(),它也将无法对其进行字符串化.

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

Answer 4

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

Answer 5

事实上,sqlalchemy从不将数据与您的查询混合在一起.查询和数据分别传递给底层数据库驱动程序 - 数据插值发生在数据库中.

Sqlalchemy将您查看的查询传递str(myquery)给数据库,值将在单独的元组中传递.

你可以使用一些方法,你自己用查询插入数据(如albertov建议的那样),但这与sqlalchemy执行的不同.

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

Answer 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，您可以将 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 )

Answer 8

首先，让我先说一下，我假设您这样做主要是出于调试目的-我不建议您尝试在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()) # reasonably 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。