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

2020/12/02 18:11 · python ·  · 0评论

我有一个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的一般性质。我不在乎查询是否在我发现查询之前就已经执行了;我只是想知道!

博客提供了更新的答案。

从博客文章中引用,建议并为我工作。

>>> 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的回答!希望对其他在这里搜索的人有所帮助。

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

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

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

该文档还发出以下警告:

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

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

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

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

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

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

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

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

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

您可以使用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)
...

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

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

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

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

我已经创建了这个小功能,当我想打印完整查询时会导入该功能,考虑到当方言已经绑定时我正处于测试中间:

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
本文地址:http://python.askforanswer.com/ruhecongsqlalchemybiaodashihuoquyuanshidebianyisqlchaxun.html
文章标签: ,   ,   ,  
版权声明:本文为原创文章,版权归 admin 所有,欢迎分享本文,转载请保留出处!

文件下载

老薛主机终身7折优惠码boke112

上一篇:
下一篇:

评论已关闭!