Bagaimana cara mendapatkan kueri SQL mentah yang dikompilasi dari ekspresi SQLAlchemy?

Saya memiliki objek kueri SQLAlchemy dan ingin mendapatkan teks dari pernyataan SQL yang dikompilasi, dengan semua parameternya terikat (misalnya tidak ada %satau variabel lain yang menunggu untuk diikat oleh penyusun pernyataan atau mesin dialek MySQLdb, dll).

Memanggil str()kueri akan mengungkapkan sesuatu seperti ini:

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

Saya sudah mencoba mencari di query._params tetapi itu adalah dikt yang kosong. Saya menulis kompiler saya sendiri menggunakan contoh sqlalchemy.ext.compiler.compilesdekorator ini tetapi bahkan pernyataan di sana masih memiliki %stempat yang saya inginkan datanya.

Saya tidak tahu kapan parameter saya dicampur untuk membuat kueri; saat memeriksa objek kueri, mereka selalu berupa kamus kosong (meskipun kueri dijalankan dengan baik dan mesin mencetaknya saat Anda mengaktifkan log gema).

Saya mulai mendapatkan pesan bahwa SQLAlchemy tidak ingin saya mengetahui kueri yang mendasarinya, karena hal itu merusak sifat umum antarmuka API ekspresi semua DB-API yang berbeda. Saya tidak keberatan jika kueri dieksekusi sebelum saya menemukan apa itu; Saya hanya ingin tahu!

Blog ini memberikan jawaban yang diperbarui.

Mengutip dari posting blog, ini disarankan dan berhasil untuk saya.

>>> from sqlalchemy.dialects import postgresql
>>> print str(q.statement.compile(dialect=postgresql.dialect()))

Dimana q didefinisikan sebagai:

>>> q = DBSession.query(model.Name).distinct(model.Name.value) \
             .order_by(model.Name.value)

Atau sembarang jenis session.query ().

Terima kasih kepada Nicolas Cadou atas jawabannya! Saya harap ini membantu orang lain yang datang mencari di sini.

The dokumentasi menggunakan literal_bindsuntuk mencetak query qtermasuk parameter:

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

pendekatan di atas memiliki peringatan bahwa itu hanya didukung untuk tipe dasar, seperti int dan string, dan lebih jauh lagi jika bindparam () tanpa nilai pra-setel digunakan secara langsung, itu juga tidak akan dapat merangkai itu.

Dokumentasi juga mengeluarkan peringatan ini:

Jangan pernah menggunakan teknik ini dengan konten string yang diterima dari input yang tidak tepercaya, seperti dari formulir web atau aplikasi input pengguna lainnya. Fasilitas SQLAlchemy untuk memaksa nilai Python menjadi nilai string SQL langsung tidak aman terhadap input yang tidak tepercaya dan tidak memvalidasi jenis data yang diteruskan. Selalu gunakan parameter terikat saat secara terprogram memanggil pernyataan SQL non-DDL terhadap database relasional.

Ini harus bekerja dengan 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)

Untuk backend MySQLdb saya memodifikasi jawaban albertov yang luar biasa (terima kasih banyak!) Sedikit. Saya yakin mereka dapat digabungkan untuk memeriksa apakah comp.positionalada Truetetapi itu sedikit di luar cakupan pertanyaan ini.

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)

Masalahnya, sqlalchemy tidak pernah mencampur data dengan kueri Anda. Kueri dan data diteruskan secara terpisah ke driver database yang mendasarinya - interpolasi data terjadi di database Anda.

Sqlalchemy meneruskan kueri seperti yang Anda lihat di str(myquery)database, dan nilainya akan ditempatkan dalam tupel terpisah.

Anda dapat menggunakan beberapa pendekatan di mana Anda menginterpolasi data dengan kueri sendiri (seperti yang disarankan albertov di bawah), tetapi itu bukan hal yang sama yang dijalankan oleh sqlalchemy.

Pertama izinkan saya memberi pengantar dengan mengatakan bahwa saya berasumsi Anda melakukan ini terutama untuk tujuan debugging - Saya tidak akan merekomendasikan mencoba mengubah pernyataan di luar SQLAlchemy fluent API.

Sayangnya, tampaknya tidak ada cara sederhana untuk menampilkan pernyataan yang dikompilasi dengan parameter kueri yang disertakan. SQLAlchemy sebenarnya tidak memasukkan parameter ke dalam pernyataan - mereka diteruskan ke mesin database sebagai kamus . Ini memungkinkan pustaka khusus database menangani hal-hal seperti keluar dari karakter khusus untuk menghindari injeksi SQL.

Tetapi Anda dapat melakukan ini dalam proses dua langkah dengan cukup mudah. Untuk mendapatkan pernyataan tersebut, Anda dapat melakukan seperti yang telah Anda tunjukkan, dan cukup mencetak kueri:

>>> print(query)
SELECT field_1, field_2 FROM table WHERE id=%s;

Anda bisa selangkah lebih dekat dengan query.statement, untuk melihat nama parameter. Catatan di :id_1bawah vs di %satas - sebenarnya bukan masalah dalam contoh yang sangat sederhana ini, tetapi bisa menjadi kunci dalam pernyataan yang lebih rumit.

>>> 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;

Kemudian, Anda bisa mendapatkan nilai parameter yang sebenarnya dengan mendapatkan paramsproperti dari pernyataan yang dikompilasi:

>>> print(query.statement.compile().params)
{u'id_1': 1} 

Ini bekerja setidaknya untuk backend MySQL; Saya berharap ini juga cukup umum untuk PostgreSQL tanpa perlu digunakan psycopg2.

Untuk backend postgresql menggunakan psycopg2, Anda dapat mendengarkan do_executeacara, kemudian menggunakan cursor, pernyataan, dan ketik parameter yang dipaksa bersama dengan Cursor.mogrify()untuk menyebariskan parameter. Anda dapat mengembalikan True untuk mencegah eksekusi kueri yang sebenarnya.

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

Penggunaan sampel:

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

Solusi berikut menggunakan SQLAlchemy Expression Language dan bekerja dengan SQLAlchemy 1.1. Solusi ini tidak mencampur parameter dengan kueri (seperti yang diminta oleh penulis asli), tetapi menyediakan cara menggunakan model SQLAlchemy untuk menghasilkan string kueri SQL dan kamus parameter untuk dialek SQL yang berbeda. Contohnya berdasarkan tutorial http://docs.sqlalchemy.org/en/rel_1_0/core/tutorial.html

Mengingat kelasnya,

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

kami dapat menghasilkan pernyataan kueri menggunakan fungsi pilih .

from sqlalchemy.sql import select    
statement = select([foo.name, foo.value]).where(foo.value > 0)

Selanjutnya, kita dapat mengompilasi pernyataan tersebut menjadi objek kueri.

query = statement.compile()

Secara default, pernyataan dikompilasi menggunakan implementasi 'bernama' dasar yang kompatibel dengan database SQL seperti SQLite dan Oracle. Jika Anda perlu menentukan dialek seperti PostgreSQL, Anda bisa melakukannya

from sqlalchemy.dialects import postgresql
query = statement.compile(dialect=postgresql.dialect())

Atau jika Anda ingin secara eksplisit menetapkan dialek sebagai SQLite, Anda dapat mengubah paramstyle dari 'qmark' menjadi 'bernama'.

from sqlalchemy.dialects import sqlite
query = statement.compile(dialect=sqlite.dialect(paramstyle="named"))

Dari objek query, kita dapat mengekstrak string query dan parameter query

query_str = str(query)
query_params = query.params

dan terakhir jalankan kueri.

conn.execute( query_str, query_params )

Anda dapat menggunakan acara dari keluarga ConnectionEvents : after_cursor_executeatau before_cursor_execute.

Dalam sqlalchemy UsageRecipes oleh @zzzeek Anda dapat menemukan contoh ini:

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

Di sini Anda bisa mendapatkan akses ke pernyataan Anda

Jadi, mengumpulkan banyak potongan kecil dari jawaban yang berbeda ini, saya menemukan apa yang saya butuhkan: satu set kode sederhana untuk dimasukkan dan kadang-kadang tetapi dapat diandalkan (yaitu menangani semua tipe data) ambil SQL terkompilasi yang tepat yang dikirim ke my Postgres backend dengan hanya menginterogasi kueri itu sendiri:

from sqlalchemy.dialects import postgresql

query = [ .... some ORM query .... ]

compiled_query = query.statement.compile(
    dialect=postgresql.dialect()
)
mogrified_query = session.connection().connection.cursor().mogrify(
    str(compiled_query),
    compiled_query.params
)

print("compiled SQL = {s}".format(mogrified_query.decode())

Saya pikir .statement mungkin akan berhasil: 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