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

У меня есть объект запроса 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 decorator, но даже в заявлении все еще есть %s где я хочу данные.

Я не могу понять, когда мои параметры смешиваются для создания запроса; при исследовании объекта запроса они всегда являются пустым словарем (хотя запрос выполняется отлично, и двигатель выводит его при включении эхо-журнала).

Я начинаю получать сообщение о том, что SQLAlchemy не хочет, чтобы я знал базовый запрос, поскольку он нарушает общий характер интерфейса API выражения всех разных DB-API. Я не возражаю, если запрос выполняется, прежде чем я узнаю, что это такое; Я просто хочу знать!

  • sqlalchemy полиморфные многие-ко-многим
  • Как я могу профилировать приложение с поддержкой SQLAlchemy?
  • Обратный кадр данных Pandas из запроса PostgreSQL с sqlalchemy
  • SqlAlchemy конвертирует UTC DateTime в локальное время перед сохранением
  • сигналов или триггеров в SQLAlchemy
  • нужна проверка и добавление базы данных sqlite для ведения домашнего хозяйства и управления кодом
  • SQLAlchemy DateTime часовой пояс
  • Как элегантно проверить существование объекта / экземпляра / переменной и одновременно назначить его переменной, если она существует в python?
  • 9 Solutions collect form web for “Как получить исходный, скомпилированный 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 ) 
    Python - лучший язык программирования в мире.