Событие before_cursor_execute

Событие before_cursor_execute срабатывает перед тем, как курсор базы данных выполнит SQL-запрос. Обработчик события принимает соединение, курсор, строку SQL-запроса и параметры запроса. Событие позволяет логировать запросы, замерять время их выполнения или модифицировать SQL перед отправкой в базу данных.

Первым параметром обработчик получает объект соединения conn, вторым - объект курсора cursor, третьим - строку SQL-запроса statement, четвертым - кортеж параметров parameters, пятым - контекст выполнения context, шестым - флаг executemany.

Синтаксис

from sqlalchemy import event

@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	pass

Пример

Давайте подпишемся на событие и выведем SQL-запрос перед его выполнением:

from sqlalchemy import create_engine, event, text

engine = create_engine('sqlite:///:memory:')

@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	print(statement)

with engine.connect() as conn:
	conn.execute(text('SELECT 1'))

Итоговый SQL запрос:

SELECT 1

Пример

Давайте замерим время выполнения каждого запроса с помощью события before_cursor_execute:

import time
from sqlalchemy import create_engine, event, text

engine = create_engine('sqlite:///:memory:')
timings = {}

@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	conn.info['query_start_time'] = time.time()

@event.listens_for(engine, 'after_cursor_execute')
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	elapsed = time.time() - conn.info['query_start_time']
	timings[statement] = elapsed

with engine.connect() as conn:
	conn.execute(text('SELECT 1'))

print(timings)

Результат выполнения кода:

{'SELECT 1': 0.000123456789}

Пример

Давайте будем записывать все выполненные запросы в список с помощью события before_cursor_execute:

from sqlalchemy import create_engine, event, text

engine = create_engine('sqlite:///:memory:')
queries = []

@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	queries.append(statement)

with engine.connect() as conn:
	conn.execute(text('SELECT 1'))
	conn.execute(text('SELECT 2'))

print(queries)

Результат выполнения кода:

['SELECT 1', 'SELECT 2']

Смотрите также

событие after_cursor_execute,
которое срабатывает после выполнения запроса
функцию listens_for,
которая подписывает обработчик на событие
функцию listen,
которая регистрирует обработчик события
функцию remove,
которая удаляет обработчик события