使用SQLAlchemy构建有序一对多关系及级联删除策略

class Folder(Base):
    __tablename__ = "folder"
    id = Column(Integer, primary_key=True)
    items = relationship(
        "Item",
        back_populates="folder",
        cascade="all, delete-orphan",
    )
    item_ordering = Column(ARRAY(String), default=[]) # 存储ID列表

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

Base = declarative_base()

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

    # 通过关联对象访问项目，并管理级联删除
    item_associations = relationship(
        "FolderItemAssociation",
        back_populates="folder",
        order_by="desc(FolderItemAssociation.order)", # 按order字段降序排列
        single_parent=True, # 确保关联对象只属于一个父对象
        cascade="all, delete-orphan", # 级联删除关联对象
    )

    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"

    project_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，从而实现逻辑上的1:N
    )
    order = Column(
        BigInteger,
        autoincrement=True, # 尝试自动递增，但可能需要手动管理以确保顺序可控
    )

    folder = relationship(
        "Folder",
        back_populates="item_associations",
    )
    item = relationship(
        "Item",
        back_populates="folder_association",
        # 关键：当关联记录被删除时，级联删除对应的Item
        cascade="all, delete-orphan",
        single_parent=True # 确保Item只通过此关联被拥有
    )

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

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

    # Item通过关联对象反向引用其所属的Folder
    folder_association = relationship(
        "FolderItemAssociation",
        back_populates="item",
        passive_deletes=True, # 优化删除性能，允许数据库处理级联删除
        uselist=False, # Item只与一个关联对象关联
    )

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

import sys
from sqlalchemy import create_engine, Integer, String, BigInteger
from sqlalchemy.schema import Column, ForeignKey
from sqlalchemy.orm import Session

# 省略模型定义，同上文

# 配置数据库连接 (这里使用PostgreSQL为例，请根据实际情况修改)
# username, password, db = sys.argv[1:4] # 从命令行参数获取
# engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=False)
engine = create_engine("sqlite:///:memory:", echo=False) # 为简化示例，使用内存SQLite

Base.metadata.create_all(engine) # 创建所有表

def reset(session):
    """清空所有表数据"""
    session.query(Folder).delete()
    session.query(FolderItemAssociation).delete()
    session.query(Item).delete()
    session.commit()
    assert_counts(session, (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):
    """获取当前表中的记录数量"""
    return (
        session.query(Folder).count(),
        session.query(FolderItemAssociation).count(),
        session.query(Item).count(),
    )

def create_sample_folders(session):
    """创建示例数据"""
    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(f"Created: {repr(folder1)}")
    print(f"Created: {repr(folder2)}")

def test_folder_deletion_cascades_to_items():
    """测试删除文件夹时，项目及其关联记录是否被级联删除"""
    with Session(engine) as session:
        reset(session) # 确保初始状态为空
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4)) # 2个文件夹，4个关联，4个项目

        # 删除第一个文件夹
        folder_to_delete = session.query(Folder).first()
        session.delete(folder_to_delete)
        session.commit()

        # 验证剩余数量：1个文件夹，2个关联，2个项目
        assert_counts(session, (1, 2, 2))

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

def test_item_deletion_does_not_delete_folder():
    """测试删除项目时，文件夹是否不被删除"""
    with Session(engine) as session:
        reset(session)
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4))

        # 删除一个项目
        item_to_delete = session.query(Item).first()
        session.delete(item_to_delete)
        session.commit()

        # 验证剩余数量：2个文件夹，3个关联，3个项目
        # 注意：这里删除Item会级联删除其关联记录，但不会影响其他Item或Folder
        assert_counts(session, (2, 3, 3))

        reset(session)

def test_association_deletion_cascades_to_item_only():
    """测试删除关联记录时，仅级联删除项目，不影响文件夹"""
    with Session(engine) as session:
        reset(session)
        create_sample_folders(session)
        assert_counts(session, (2, 4, 4))

        # 删除一个关联记录
        assoc_to_delete = session.query(FolderItemAssociation).first()
        session.delete(assoc_to_delete)
        session.commit()

        # 验证剩余数量：2个文件夹，3个关联，3个项目
        # 删除关联记录会级联删除对应的Item
        assert_counts(session, (2, 3, 3))

        reset(session)

# 运行所有测试
test_folder_deletion_cascades_to_items()
test_item_deletion_does_not_delete_folder()
test_association_deletion_cascades_to_item_only()

print("\n所有测试通过！")