Исключение CircularDependencyError
Исключение CircularDependencyError наследуется от
FlushError
и возникает во время операции flush, когда SQLAlchemy
не может определить порядок вставки строк из-за циклических
зависимостей между внешними ключами. Такая ситуация появляется,
когда две или более таблиц ссылаются друг на друга через
внешние ключи, и ни одну из них нельзя вставить первой.
Исключение сигнализирует о том, что автоматическое
упорядочивание вставки невозможно и требуется ручное
разрешение конфликта.
Синтаксис
from sqlalchemy.exc import CircularDependencyError
Пример
Давайте создадим две модели с взаимными внешними ключами и попробуем сохранить их в сессии:
from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
status_id: Mapped[int] = mapped_column(ForeignKey('statuses.id'))
status: Mapped['Status'] = relationship(back_populates='articles')
class Status(Base):
__tablename__ = 'statuses'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
article_id: Mapped[int] = mapped_column(ForeignKey('articles.id'))
articles: Mapped[list['Article']] = relationship(back_populates='status')
Теперь попробуем выполнить операцию flush с объектами,
которые ссылаются друг на друга:
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.exc import CircularDependencyError
from models import Base, Article, Status
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
article = Article(id=1, title='article')
status = Status(id=1, name='user')
article.status = status
status.articles = [article]
session.add(article)
session.add(status)
try:
session.flush()
except CircularDependencyError as err:
print(f"Error: {err}")
Результат выполнения кода:
"Error: Circular dependency detected: articles -> statuses -> articles"
Пример
Для разрешения циклической зависимости можно использовать
параметр post_update в функции relationship.
Это заставит SQLAlchemy выполнить дополнительный UPDATE-запрос
после первоначальной вставки. Перепишем модель:
from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
status_id: Mapped[int] = mapped_column(ForeignKey('statuses.id'))
status: Mapped['Status'] = relationship(
back_populates='articles',
post_update=True
)
class Status(Base):
__tablename__ = 'statuses'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
article_id: Mapped[int] = mapped_column(ForeignKey('articles.id'))
articles: Mapped[list['Article']] = relationship(back_populates='status')
Теперь операция flush завершится успешно, а SQLAlchemy
выполнит дополнительный UPDATE-запрос:
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import Base, Article, Status
engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
article = Article(id=1, title='article')
status = Status(id=1, name='user')
article.status = status
status.articles = [article]
session.add(article)
session.add(status)
session.flush()
print('Flush completed successfully')
Итоговый SQL запрос:
INSERT INTO statuses (id, name, article_id) VALUES (1, 'user', NULL)
INSERT INTO articles (id, title, status_id) VALUES (1, 'article', 1)
UPDATE statuses SET article_id = 1 WHERE statuses.id = 1
Смотрите также
-
исключение
FlushError,
базовый класс для ошибок операции flush -
исключение
StaleDataError,
которое возникает при конфликте версий данных -
исключение
IntegrityError,
которое возникает при нарушении целостности данных -
исключение
SQLAlchemyError,
базовый класс всех исключений SQLAlchemy