How do I get a raw, compiled SQL query from a SQLAlchemy expression?_问答_开发者

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 ge开发者_运维百科ts executed before I found out what it was; I just want to know!

This blogpost by Nicolas Cadou 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().

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.

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)

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.

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)

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.

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

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 )

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

UPDATE: Came up with yet another case where the previous solution here wasn't properly producing the correct SQL statement. After a bit of diving around in SQLAlchemy, it becomes apparent that you not only need to compile for a particular dialect, you also need to take the compiled query and initialize it for the correct DBAPI connection context. Otherwise, things like type bind processors don't get executed and values like JSON.NULL don't get properly translated.

Note, this makes this solution very particular to Flask + Flask-SQLAlchemy + psycopg2 + PostgreSQL. You may need to translate this solution to your environment by changing the dialect and how you reference your connection. However, I'm pretty confident this produces the exact SQL for all data types.

The result below is a simple method to drop in and occasionally but reliably grab the exact, compiled SQL that would be sent to my PostgreSQL backend by just interrogating the query itself:

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

I've created this little function that I import when I want to print the full query, considering I'm in the middle of a test when the dialect is already bound:

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

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

If with SQLAlchemy you are using PyMySQL, you can do one trick.

I was in a hurry and lost a lot of time, so I changed the driver for print the current statement with parameters.

SQLAlchemy intentionally does not support full stringification of literal values.

But PyMySQL has 'mogrify' method which does it, but, SQLALchemy has no HOOK for call it when using ORM insert/update (when it controls the cursor) like db.add or commit/flush (for update).

So, Just go where the driver is using (to know where use): pip show pycharm

In the folder, find and edit the file cursors.py.

In the method: