SQLAlchemy与SQLite：解决外键级联删除失效问题

from sqlalchemy.engine import Engine
from sqlalchemy import event

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    """
    为每个新的SQLite连接启用外键约束。
    """
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

from sqlalchemy import create_engine, Column, Integer, ForeignKey, String
from sqlalchemy.orm import relationship, sessionmaker, declarative_base, Mapped, mapped_column
from sqlalchemy.engine import Engine
from sqlalchemy import event

# 1. 配置SQLite连接时启用外键约束
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    """
    为每个新的SQLite连接启用外键约束。
    """
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

# 2. 数据库引擎和会话设置
engine = create_engine(
    "sqlite:///./database.db",
    echo=True,  # 开启echo可以观察SQL语句执行情况
)
Session = sessionmaker(bind=engine)
Base = declarative_base()

# 3. 模型定义
class UserOrm(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(nullable=True, default="")
    age: Mapped[int] = mapped_column(nullable=True, default=0)

    # 配置关系，使用cascade="all, delete"和passive_deletes=True
    # passive_deletes=True 告诉ORM，当父对象删除时，不要加载子对象来执行删除，
    # 而是依赖数据库的外键级联删除。这与ondelete="CASCADE"配合使用。
    computers = relationship(
        "ComputerOrm",
        back_populates="host",
        cascade="all, delete",  # ORM层面的级联删除（针对session.delete()）
        passive_deletes=True,   # 配合数据库的ON DELETE CASCADE
    )

class ComputerOrm(Base):
    __tablename__ = "computers"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    # 配置外键，并设置ondelete="CASCADE"
    host_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), index=True, nullable=False)
    host = relationship(UserOrm, back_populates="computers")

# 4. 创建表结构
Base.metadata.create_all(engine)

# 5. 插入测试数据
def setup_data():
    with Session() as session:
        # 清理旧数据
        session.query(ComputerOrm).delete()
        session.query(UserOrm).delete()
        session.commit()

        user1 = UserOrm(name="Alice", age=30)
        user2 = UserOrm(name="Bob", age=25)

        computer1 = ComputerOrm(name="Alice_PC", host=user1)
        computer2 = ComputerOrm(name="Alice_Laptop", host=user1)
        computer3 = ComputerOrm(name="Bob_Workstation", host=user2)

        session.add_all([user1, user2, computer1, computer2, computer3])
        session.commit()
        print(f"Added User: {user1.id}, {user1.name}")
        print(f"Added User: {user2.id}, {user2.name}")
        return user1.id, user2.id

# 6. 执行删除操作
def perform_delete(user_id_to_delete):
    with Session() as session:
        print(f"\n--- 删除用户 ID: {user_id_to_delete} ---")
        # 删除前查询数量
        initial_users = session.query(UserOrm).count()
        initial_computers = session.query(ComputerOrm).count()
        print(f"删除前：用户数 = {initial_users}, 电脑数 = {initial_computers}")

        # 使用query().delete()进行批量删除
        session.query(UserOrm).filter_by(id=user_id_to_delete).delete()
        session.commit()

        # 删除后查询数量
        final_users = session.query(UserOrm).count()
        final_computers = session.query(ComputerOrm).count()
        print(f"删除后：用户数 = {final_users}, 电脑数 = {final_computers}")

        # 验证是否级联删除成功
        if initial_users > final_users and initial_computers > final_computers:
            print(f"用户 ID {user_id_to_delete} 及其关联的电脑已成功级联删除。")
        else:
            print(f"用户 ID {user_id_to_delete} 删除成功，但关联电脑级联删除失败或无关联电脑。")

if __name__ == "__main__":
    user1_id, user2_id = setup_data()
    perform_delete(user1_id)
    perform_delete(user2_id)

    # 再次验证所有数据是否被清理
    with Session() as session:
        print(f"\n--- 最终数据状态 ---")
        print(f"用户数 = {session.query(UserOrm).count()}")
        print(f"电脑数 = {session.query(ComputerOrm).count()}")