如何从SQLAlchemyexpression式获取原始的编译SQL查询？
我有一个SQLAlchemy查询对象，并希望得到编译的SQL语句的文本，其所有参数绑定（例如没有%s或其他variables等待被语句编译器或MySQLdb方言引擎等绑定）。
在查询中调用str()显示如下所示的内容：
SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC
我试过查找query._params，但它是一个空字典。 我使用sqlalchemy.ext.compiler.compiles修饰器的这个例子编写了我自己的编译器，但是即使这个语句仍然有我想要数据的%s 。
我不能完全弄清楚什么时候我的参数混入来创build查询; 当检查查询对象时，它们总是一个空的字典（尽pipe查询执行正常，当您打开回显logging时，引擎会将其打印出来）。
我开始收到SQLAlchemy不希望我知道底层查询的消息，因为它打破了expression式API接口的所有不同DB-API的一般性质。 我不介意在查询之前执行查询是什么; 我只是想知道！
这应该与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)
此博客提供了一个更新的答案。
从博客引用，这是build议和为我工作。
>>> 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的回答！ 我希望能帮助其他人来这里search。
该文档使用literal_binds打印包含参数的查询q ：
print(q.statement.compile(compile_kwargs={"literal_binds": True}))
上面的方法有一点要注意，它只支持基本的types，例如整数和string，而且如果直接使用没有预设值的bindparam（），它将不能够对其进行string化。
对于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)
事情是，sqlalchemy从不混合您的查询的数据。 查询和数据分别传递给你的下划线数据库驱动程序 – 数据插值发生在你的数据库中。
sqlalchemy将你在str(myquery)看到的查询传递给数据库，这些值将在一个单独的元组中进行。
你可以使用一些你自己插入数据和查询的方法（如下面的albertovbuild议的那样），但这与sqlalchemy正在执行的不同。
对于使用psycopg2的postgresql后端，您可以侦听do_execute事件，然后使用游标，语句和types强制参数以及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"}}'}
您可以使用来自ConnectionEvents族的事件： after_cursor_execute或before_cursor_execute 。
在@zzzeek的sqlalchemy UsageRecipes中，你可以find这个例子：
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) ...
在这里，您可以访问您的声明
我认为.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
以下解决scheme使用SQLAlchemyexpression式语言并与SQLAlchemy 1.1配合使用。 此解决scheme不会将参数与查询混合（按照原始作者的要求），但提供了一种使用SQLAlchemy模型为不同的SQL方言生成SQL查询string和参数字典的方法。 该示例基于教程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()
默认情况下，语句是使用与SQL数据库（如SQLite和Oracle）兼容的基本“命名”实现进行编译的。 如果你需要指定一个方言，如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"))
从查询对象中，我们可以提取查询string和查询参数
query_str = str(query) query_params = query.params
最后执行查询。
conn.execute( query_str, query_params )
