SQLAlchemy 식에서 컴파일 된 원시 SQL 쿼리를 가져 오는 방법은 무엇입니까?

SQLAlchemy 식에서 컴파일 된 원시 SQL 쿼리를 가져 오는 방법은 무엇입니까?

---

SQLAlchemy 쿼리 개체가 있고 모든 매개 변수가 바인딩 된 컴파일 된 SQL 문의 텍스트를 가져오고 싶습니다 (예 : %s문 컴파일러 또는 MySQLdb 언어 엔진에 의해 바인딩되기를 기다리는 변수 가 없거나 다른 변수 등).

str()쿼리를 호출 하면 다음과 같은 결과가 나타납니다.

SELECT id WHERE date_added <= %s AND date_added >= %s ORDER BY count DESC

query._params에서 찾아 봤지만 빈 dict입니다. 이 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}))

위의 접근 방식은 정수 및 문자열과 같은 기본 유형에 대해서만 지원되며 사전 설정된 값이없는 bindparam ()을 직접 사용하는 경우에도이를 문자열화할 수 없다는 경고가 있습니다.

문서는 또한 다음 경고를 발행합니다.

웹 양식 또는 기타 사용자 입력 응용 프로그램에서와 같이 신뢰할 수없는 입력에서받은 문자열 콘텐츠에는이 기술을 사용하지 마십시오. Python 값을 직접 SQL 문자열 값으로 강제 변환하는 SQLAlchemy의 기능은 신뢰할 수없는 입력에 대해 안전하지 않으며 전달되는 데이터 유형의 유효성을 검사하지 않습니다. 관계형 데이터베이스에 대해 비 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)

---

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)

---

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

---

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()) # reasonably 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.

---

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

---

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

참고URL : https://stackoverflow.com/questions/4617291/how-do-i-get-a-raw-compiled-sql-query-from-a-sqlalchemy-expression