Исключение TimeoutError

Исключение TimeoutError в SQLAlchemy возникает, когда операция с базой данных превышает заданное время ожидания. Оно наследуется от класса SQLAlchemyError и является частью группы ошибок уровня СУБД. Обычно исключение выбрасывается при блокировке таблиц, недоступности сервера или превышении таймаута пула соединений.

Синтаксис

from sqlalchemy.exc import TimeoutError

TimeoutError(statement, params, orig)

Исключение принимает необязательные параметры statement, params и orig, которые содержат SQL-запрос, его параметры и исходную ошибку драйвера соответственно.

Пример

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

from sqlalchemy import create_engine
from sqlalchemy.exc import TimeoutError

engine = create_engine(
	'sqlite:///:memory:',
	pool_size=1,
	pool_timeout=0
)

conn1 = engine.connect()

try:
	conn2 = engine.connect()
except TimeoutError as exc:
	print('TimeoutError:', exc)

conn1.close()

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

"TimeoutError: QueuePool limit of size 1 overflow 0 reached, connection timed out, timeout 0.00"

Пример

Давайте выполним запрос к таблице articles с малым таймаутом блокировки и посмотрим на обработку исключения:

from sqlalchemy import create_engine, text
from sqlalchemy.exc import TimeoutError

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

with engine.connect() as conn:
	conn.execute(text('CREATE TABLE articles (id INTEGER, title TEXT)'))
	conn.execute(text("INSERT INTO articles VALUES (1, 'article')"))
	conn.commit()
	
	try:
		res = conn.execute(
			text('SELECT * FROM articles'),
			execution_options={'timeout': 0}
		)
		print(res.fetchall())
	except TimeoutError as exc:
		print('Statement timeout:', exc)

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

SELECT * FROM articles

Пример

Давайте проверим, как исключение обрабатывается вместе с базовым классом SQLAlchemyError:

from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError, TimeoutError

engine = create_engine(
	'sqlite:///:memory:',
	pool_size=1,
	pool_timeout=0
)

conn1 = engine.connect()

try:
	conn2 = engine.connect()
except SQLAlchemyError as exc:
	print('Caught as SQLAlchemyError')
	print('Is TimeoutError:', isinstance(exc, TimeoutError))

conn1.close()

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

"Caught as SQLAlchemyError"
"Is TimeoutError: True"

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

класс SQLAlchemyError,
базовый класс всех исключений SQLAlchemy
класс DBAPIError,
обертка над ошибками драйвера базы данных
класс DisconnectionError,
исключение при разрыве соединения с базой
класс OperationalError,
исключение при ошибках работы СУБД