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

小开

这应该可以用于 Sql 炼金术 > = 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)

小开

问题是，sql 炼金术从来不会把数据和查询混在一起。查询和数据分别传递给底层数据库驱动程序-数据的插值发生在数据库中。

正如您在 str(myquery)中看到的那样，Sqlchemy 将查询传递给数据库，这些值将放在一个单独的元组中。

您可以使用某种方法，自己用查询来插入数据(如下面的 albertov 所建议的) ，但是这与 sql 炼金术执行的内容不同。

小开

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

小开

最佳答案

这个博客提供了一个更新的答案。

引用博客文章中的话，这个建议对我来说是有效的。

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

感谢尼古拉斯卡杜的回答! 我希望它能帮助其他人来这里寻找。

小开

我认为. 陈述可能会起作用: 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

小开

您可以使用来自 ConnectionEvents系列的事件: after_cursor_execute或 before_cursor_execute。

在@zzzeek 的 sql伟力用法食谱中，你可以找到这样一个例子:

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

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

小开

文件使用 literal_binds打印一个包含参数的 q查询:

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

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

该文件还发出了这样的警告:

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

小开

下面的解决方案使用 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())

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

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 )

小开

对于使用 Psycopg2的 postgreql 后端，您可以侦听 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"}}'}

小开

首先，我假设您这样做主要是为了调试目的——我不建议尝试修改 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。

小开

更新: 提出了另一种情况，即以前的解决方案没有正确地生成正确的 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)}")

小开

我已经创建了这个小函数，当我想打印完整的查询时，我会导入它，考虑到我正在进行一个方言已经绑定的测试:

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"\nPrinting SQLAlchemy query:\n\n")
print(sql)
return sql

小开

如果使用 SQLAlchemy 时使用的是 PyMySQL，那么可以使用一个技巧。

我很匆忙，浪费了很多时间，所以我更改了驱动程序，以便用参数打印当前语句。

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

但是 PyMySQL 有一个“ mogrify”方法可以做到这一点，但是 SQLALchemy 在使用 ORM 插入/更新(当它控制游标时)时没有 HOOK 来调用它，比如 db.add 或提交/刷新(用于更新)。

所以，只要去驱动程序正在使用的地方(知道在哪里使用) : 皮普秀魅力

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

方法:

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

在这条线下:

query = self.mogrify(query, args)

只需添加:

print(query)

将工作像一个魅力，调试，解决问题，并删除打印。