Как получить исходный, скомпилированный SQL-запрос из выражения SQLAlchemy?

Это должно работать с 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) 

Этот блог содержит обновленный ответ.

Цитируя из сообщения в блоге, это предлагается и работает для меня.

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

вышеупомянутый подход имеет оговорки, что он поддерживается только для базовых типов, таких как ints и strings, и, кроме того, если bindparam () без предварительно заданного значения используется напрямую, он также не сможет это сделать.

Для бэкэнда MySQLdb я изменил удивительный ответ Альбертова (спасибо большое!) Немного. Я уверен, что они могут быть объединены, чтобы проверить, было ли 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) в базу данных, и значения будут поступать в отдельный кортеж.

Вы можете использовать некоторый подход, в котором вы интерполируете данные с запросом самостоятельно (как предложил Альбертов ниже), но это не то же самое, что выполняется sqlalchemy.

Для бэкэнда 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 

Использование образца:

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

В sqlalchemy UsageRecipes от @zzzeek вы можете найти этот пример:

 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 

В следующем решении используется язык выражения SQLAlchemy Expression Language и работает с SQLAlchemy 1.1. Это решение не смешивает параметры с запросом (в соответствии с запросом оригинального автора), но предоставляет способ использования моделей SQLAlchemy для генерации строк SQL-запросов и словарей параметров для разных диалектов SQL. Пример основан на учебнике 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() 

По умолчанию оператор компилируется с использованием базовой реализации «named», которая совместима с базами данных SQL, такими как SQLite и Oracle. Если вам нужно указать диалект, такой как PostgreSQL, вы можете сделать

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

Или, если вы хотите явно указать диалект как SQLite, вы можете изменить paramstyle с «qmark» на «named».

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

Из объекта запроса мы можем извлечь строку запроса и параметры запроса

 query_str = str(query) query_params = query.params 

и, наконец, выполнить запрос.

 conn.execute( query_str, query_params )