Событие handle_error

Событие handle_error из секции event срабатывает, когда в процессе работы соединения с базой данных возникает ошибка уровня СУБД. Это событие принадлежит движку и позволяет перехватить исключение, возникшее при выполнении запроса или во время работы с соединением. Первым параметром обработчик принимает объект исключения exception_context, который содержит информацию об ошибке, а также ссылки на движок и соединение.

Событие handle_error полезно для логирования ошибок базы данных, преобразования исключений в собственные типы, а также для реализации логики повторных попыток подключения. Обработчик может изменить атрибут retval контекста, чтобы подменить возникшее исключение на другое или полностью подавить его.

Синтаксис

from sqlalchemy import event

@event.listens_for(engine, 'handle_error')
def handle_error(exception_context):
	pass

Обработчик также можно зарегистрировать через метод listen, передав целевой объект, имя события и функцию-обработчик.

Пример

Давайте создадим движок и зарегистрируем обработчик события handle_error, который будет выводить информацию о возникшей ошибке:

from sqlalchemy import create_engine, event, text

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

@event.listens_for(engine, 'handle_error')
def handle_error(exception_context):
	print('Error caught:', exception_context.original_exception)

with engine.connect() as conn:
	conn.execute(text('SELECT * FROM missing_table'))

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

"Error caught: (sqlite3.OperationalError) no such table: missing_table"

Пример

Давайте в обработчике заменим исходное исключение на собственное, записав его в атрибут retval контекста:

from sqlalchemy import create_engine, event, text

class MyDatabaseError(Exception):
	pass

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

@event.listens_for(engine, 'handle_error')
def handle_error(exception_context):
	exception_context.retval = MyDatabaseError('database failure')

with engine.connect() as conn:
	try:
		conn.execute(text('SELECT * FROM missing_table'))
	except MyDatabaseError as e:
		print('Custom error:', e)

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

"Custom error: database failure"

Пример

Давайте в обработчике подавим исключение, установив атрибут retval в значение None, чтобы движок продолжил работу без выброса ошибки:

from sqlalchemy import create_engine, event, text

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

@event.listens_for(engine, 'handle_error')
def handle_error(exception_context):
	print('Error suppressed')
	exception_context.retval = None

with engine.connect() as conn:
	conn.execute(text('SELECT * FROM missing_table'))
	print('Connection still works')

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

"Error suppressed"
"Connection still works"

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

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