Исключение PendingRollbackError

Исключение PendingRollbackError наследуется от класса InvalidRequestError и относится к семейству SQLAlchemyError. Оно возникает, когда сессия находится в состоянии, требующем отката транзакции, но вместо этого предпринимается попытка выполнить новую операцию. Как правило, это происходит после ошибки в предыдущем запросе, которую не обработали вызовом rollback. Исключение сигнализирует, что сессия "сломана" и нуждается в явном откате для дальнейшей работы.

Синтаксис

from sqlalchemy.exc import PendingRollbackError

raise PendingRollbackError(statement, params, orig)

Пример

Давайте воспроизведем ситуацию, в которой возникает PendingRollbackError. Выполним запрос с синтаксической ошибкой, а затем попробуем сделать еще один запрос без отката:

from sqlalchemy import create_engine, text
from sqlalchemy.exc import PendingRollbackError
from sqlalchemy.orm import Session

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

with Session(engine) as session:
	try:
		session.execute(text('SELECT * FROM unknown_table'))
	except Exception as error:
		print(type(error).__name__)
	
	try:
		session.execute(text('SELECT 1'))
	except PendingRollbackError as error:
		print(type(error).__name__)

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

"OperationalError"
"PendingRollbackError"

Пример

Давайте исправим ситуацию вызовом rollback после ошибки. Тогда последующий запрос выполнится без исключения:

from sqlalchemy import create_engine, text
from sqlalchemy.exc import PendingRollbackError
from sqlalchemy.orm import Session

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

with Session(engine) as session:
	try:
		session.execute(text('SELECT * FROM unknown_table'))
	except Exception as error:
		session.rollback()
		print(type(error).__name__)
	
	try:
		res = session.execute(text('SELECT 1')).scalar()
		print(res)
	except PendingRollbackError as error:
		print(type(error).__name__)

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

"OperationalError"
1

Пример

Давайте рассмотрим случай с моделью Article, когда вставка нарушает ограничение, а следующая операция не выполняет откат:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str]
	text: Mapped[str]
	status: Mapped[str]
	num: Mapped[int]

from sqlalchemy import create_engine, text
from sqlalchemy.exc import PendingRollbackError
from sqlalchemy.orm import Session

from models import Base, Article

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
	article = Article(
		id=1,
		title='article',
		text='text',
		status='draft',
		num=1,
	)
	session.add(article)
	session.commit()
	
	duplicate = Article(
		id=1,
		title='article',
		text='text',
		status='draft',
		num=2,
	)
	session.add(duplicate)
	try:
		session.commit()
	except Exception as error:
		print(type(error).__name__)
	
	try:
		res = session.execute(text('SELECT 1')).scalar()
		print(res)
	except PendingRollbackError as error:
		print(type(error).__name__)

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

"IntegrityError"
"PendingRollbackError"

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

исключение SQLAlchemyError,
которое является базовым для всех ошибок SQLAlchemy
исключение InvalidRequestError,
от которого наследуется PendingRollbackError
исключение IntegrityError,
которое часто приводит к необходимости отката
исключение OperationalError,
которое также может вызывать сбой транзакции