Исключение IntegrityError

Исключение IntegrityError относится к группе исключений DBAPIError и возникает, когда СУБД отклоняет операцию из-за нарушения ограничений целостности данных. К таким ограничениям относятся уникальность UNIQUE, обязательность NOT NULL, внешние ключи FOREIGN KEY и проверки CHECK. Исключение обычно возникает в момент выполнения запроса или сброса сессии flush и требует отката транзакции методом rollback.

Синтаксис

from sqlalchemy.exc import IntegrityError

Пример

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

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.exc import IntegrityError

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str] = mapped_column(String(50), unique=True)

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

with Session(engine) as session:
	session.add(Article(title='article'))
	session.commit()
	
	try:
		session.add(Article(title='article'))
		session.commit()
	except IntegrityError as err:
		session.rollback()
		print('IntegrityError caught')

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

"IntegrityError caught"

Пример

Давайте выведем текст исходной ошибки СУБД через атрибут orig:

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.exc import IntegrityError

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str] = mapped_column(String(50), unique=True)

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

with Session(engine) as session:
	session.add(Article(title='article'))
	session.commit()
	
	try:
		session.add(Article(title='article'))
		session.commit()
	except IntegrityError as err:
		session.rollback()
		print(err.orig)

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

"UNIQUE constraint failed: articles.title"

Пример

Давайте поймаем ошибку при нарушении ограничения NOT NULL:

from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.exc import IntegrityError

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str] = mapped_column(String(50), nullable=False)

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

with Session(engine) as session:
	try:
		session.add(Article(title=None))
		session.commit()
	except IntegrityError as err:
		session.rollback()
		print('NOT NULL violation')

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

"NOT NULL violation"

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

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