SQLAlchemy 中带顺序的关联对象级联删除

from sqlalchemy import (
    create_engine,
    Integer,
    String,
    BigInteger,
    Column,
    ForeignKey,
)
from sqlalchemy.orm import declarative_base, Session, relationship

Base = declarative_base()

class Folder(Base):
    __tablename__ = "folder"
    id = Column(Integer, primary_key=True)

    # 通过 FolderItemAssociation 关联 Item
    item_associations = relationship(
        "FolderItemAssociation",
        back_populates="folder",
        order_by="desc(FolderItemAssociation.order)", # 按照 order 字段降序排列
        single_parent=True,  # 确保 FolderItemAssociation 实例只有一个 Folder 父级
        cascade="all, delete-orphan", # 级联删除 FolderItemAssociation
    )

    def __repr__(self):
        return f"Folder(id={self.id}, item_associations={', '.join(repr(assoc) for assoc in self.item_associations)})"


class FolderItemAssociation(Base):
    __tablename__ = "folder_item_association"

    folder_id = Column(
        Integer,
        ForeignKey("folder.id", ondelete="CASCADE"), # 数据库层级联删除
        primary_key=True,
    )
    item_id = Column(
        Integer,
        ForeignKey("item.id", ondelete="CASCADE"), # 数据库层级联删除
        primary_key=True,
        unique=True,  # 确保一个 Item 只能关联到一个 Folder，从而实现 Item 到 Folder 的 1:M 关系
    )
    order = Column(
        BigInteger,
        # autoincrement=True, # 注意：autoincrement 在某些数据库（如PostgreSQL）中可能不会自动填充此列，需手动管理
    )

    folder = relationship(
        "Folder",
        back_populates="item_associations",
    )
    item = relationship(
        "Item",
        back_populates="folder_association",
        # 关键配置：确保 Item 的生命周期由 FolderItemAssociation 管理
        cascade="all, delete-orphan",
        single_parent=True # 确保 Item 实例只有一个 FolderItemAssociation 父级
    )

    def __repr__(self):
        return f"Assoc(id={(self.folder_id, self.item_id)}, order={self.order}, item={repr(self.item)})"


class Item(Base):
    __tablename__ = "item"
    id = Column(Integer, primary_key=True)

    # Item 反向关联到 FolderItemAssociation
    folder_association = relationship(
        "FolderItemAssociation",
        back_populates="item",
        passive_deletes=True, # 依赖数据库的 ON DELETE CASCADE 行为
        uselist=False, # Item 只有一个关联对象
    )

    def __repr__(self):
        return f"Item(id={self.id})"

# 假设 engine 已创建，例如：
# engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=False)

def reset(session):
    """清空所有表数据，用于测试前初始化。"""
    session.query(Folder).delete()
    session.query(FolderItemAssociation).delete()
    session.query(Item).delete()
    session.commit()
    assert_counts(session, (0, 0, 0)) # 确保清空后计数为0

def assert_counts(session, expected_counts):
    """验证当前各表记录数是否符合预期。"""
    counts = get_counts(session)
    assert counts == expected_counts, f'Expected {expected_counts} but got {counts}'

def get_counts(session):
    """获取 Folder, FolderItemAssociation, Item 表的记录数。"""
    return (
        session.query(Folder).count(),
        session.query(FolderItemAssociation).count(),
        session.query(Item).count(),
    )

def create_sample_folders(session):
    """创建示例数据：两个 Folder，每个包含两个 Item。"""
    folder1 = Folder(
        item_associations=[
            FolderItemAssociation(item=Item()), FolderItemAssociation(item=Item())
        ]
    )
    session.add(folder1)
    folder2 = Folder(
        item_associations=[
            FolderItemAssociation(item=Item()), FolderItemAssociation(item=Item())
        ]
    )
    session.add(folder2)
    session.commit()

    print(repr(folder1))
    print(repr(folder2))

def test_folder_deletion_cascades_to_item():
    """测试：当 Folder 被删除时，其关联的 FolderItemAssociation 和 Item 都应被删除。"""
    with Session(engine) as session:
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4)) # 初始状态：2 Folder, 4 Association, 4 Item

        # 删除一个 Folder
        session.delete(session.query(Folder).first())
        session.commit()

        # 预期：剩余 1 Folder, 2 Association, 2 Item
        assert_counts(session, (1, 2, 2))

        reset(session) # 清空数据，为下一个测试准备

def test_item_deletion_does_not_delete_folder():
    """测试：当 Item 被删除时，Folder 不应被删除，但关联对象应被删除。"""
    with Session(engine) as session:
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4))

        # 删除一个 Item
        session.delete(session.query(Item).first())
        session.commit()

        # 预期：剩余 2 Folder, 3 Association, 3 Item
        assert_counts(session, (2, 3, 3))

        reset(session)

def test_association_deletion_cascades_to_item():
    """测试：当 FolderItemAssociation 被删除时，其关联的 Item 应被删除，Folder 不应被删除。"""
    with Session(engine) as session:
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4))

        # 删除一个 FolderItemAssociation
        session.delete(session.query(FolderItemAssociation).first())
        session.commit()

        # 预期：剩余 2 Folder, 3 Association, 3 Item
        assert_counts(session, (2, 3, 3))

        reset(session)

# 运行所有测试
# test_folder_deletion_cascades_to_item()
# test_item_deletion_does_not_delete_folder()
# test_association_deletion_cascades_to_item()