SQL Chemy 식에서 원시 컴파일 SQL 쿼리를 가져오려면 어떻게 해야 합니까?

SQL Chemy 식에서 원시 컴파일 SQL 쿼리를 가져오려면 어떻게 해야 합니까?

문의 를 들어 no SQL Chemy 쿼 sqlSQL 쿼((((SQL ((((((((((((((((((((((((((((((((((((((((((((((( and and and ( ( ( ( ( 를 취득하고 싶습니다.%s또는 스테이트먼트 컴파일러 또는 MySQLdb 방언 엔진에 의해 바인드되기를 기다리는 기타 변수 등).

" "str()쿼리에는 다음과 같은 내용이 표시됩니다.

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

나는 문의해 보려고 했다._params 하지만 빈 딕트입니다.나는 이 장식가의 예를 사용하여 나만의 컴파일러를 썼지만, 심지어 거기에 있는 문장도 여전히 가지고 있다.%s데이터를 원하는 곳.

파라미터가 언제 혼재하여 쿼리를 생성하는지 알 수 없습니다.쿼리 오브젝트를 조사할 때는 항상 빈 사전입니다(쿼리가 정상적으로 실행되고 에코 로깅을 켜면 엔진이 출력합니다).

SQL Chemy에서 기본 쿼리를 알고 싶지 않다는 메시지가 나타나기 시작했습니다. 이는 표현 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().

니콜라스 카두에게 답해줘서 고마워!여기로 찾아오는 사람들에게 도움이 됐으면 좋겠어요.

이 문서에서는literal_binds를 q다음 중 하나:

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

위의 접근법에는 int나 string 등의 기본 유형에서만 지원된다는 주의사항이 있습니다.또한 사전에 설정된 값이 없는 bindparam()을 직접 사용하는 경우 이 또한 문자열화할 수 없습니다.

매뉴얼에서는 다음 경고도 발행하고 있습니다.

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

중요한 건 sqalchemy는 데이터를 당신의 쿼리와 섞지 않는다는 겁니다.쿼리와 데이터는 기본 데이터베이스 드라이버에 별도로 전달됩니다.데이터 보간은 데이터베이스에서 수행됩니다.

SQLalchemy에서 볼 수 쿼리를 합니다.str(myquery)츠키다

쿼리에서 직접 데이터를 삽입하는 접근법(아래 제안)을 사용할 수 있지만 sqlalchemy가 실행하는 것과 동일하지는 않습니다.

MySQLDB 백엔드는 알베르토브의 훌륭한 답변(감사합니다!)을 조금 수정했습니다. '아예'인지 아닌지를 할 수 것 같아요.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)

먼저 이 작업은 주로 디버깅을 목적으로 하고 있다고 생각합니다.SQLChemy fluent API 이외의 문장은 수정하지 않는 것이 좋습니다.

유감스럽게도 쿼리 파라미터가 포함된 컴파일된 스테이트먼트를 표시할 수 있는 간단한 방법은 없는 것 같습니다.SQL Chemy는 실제로 파라미터는 문장에 입력하지 않습니다.데이터베이스 엔진에 사전으로 전달됩니다.이를 통해 데이터베이스별 라이브러리는 특수 문자 이스케이프 등의 작업을 처리하여 SQL 주입을 방지할 수 있습니다.

그러나 이 작업은 2단계 프로세스로 비교적 쉽게 수행할 수 있습니다.문장을 얻으려면 이미 표시된 대로 쿼리를 인쇄하면 됩니다.

>>> 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 백엔드에 입니다.포스트그레을 할 필요가 psycopg2.

사용하는 경우 postgresql psycopg2를 수 .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

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

사투리가 이미 바인딩되어 있을 때 테스트 중이기 때문에 전체 쿼리를 출력할 때 Import하는 작은 함수를 만들었습니다.

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

.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

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