如何使用SQLAlchemy Core实现级联删除？

问题背景 #

我有两张表，user_data_relation（多端）与data（一端）为多对一关系，使用SQLAlchemy Core创建表的代码如下：

meta = MetaData()

Data = Table(
    "data", meta,
    Column('id', Integer, primary_key=True, unique=True),
    Column('data_id', String(100)),
    Column('text', String(25000)),
    Column('created_at', DateTime(timezone=True)),
)

User_Data_Relation = Table(
    "user_data_relation", meta,
    Column('id', Integer, primary_key=True),
    Column('data_id', Integer, ForeignKey('data.id')),
)

meta.create_all(engine)

由于user_data_relation表的外键约束，无法直接删除data表中的行，需要通过SQLAlchemy Core实现级联删除。

解决方案 #

1. 创建表时指定级联规则 #

在定义外键列时，通过ForeignKey的ondelete参数设置CASCADE，让数据库自动处理级联删除逻辑。修改后的表定义代码如下：

meta = MetaData()

Data = Table(
    "data", meta,
    Column('id', Integer, primary_key=True, unique=True),
    Column('data_id', String(100)),
    Column('text', String(25000)),
    Column('created_at', DateTime(timezone=True)),
)

User_Data_Relation = Table(
    "user_data_relation", meta,
    Column('id', Integer, primary_key=True),
    # 添加上ondelete="CASCADE"参数
    Column('data_id', Integer, ForeignKey('data.id', ondelete="CASCADE")),
)

meta.create_all(engine)

设置ondelete="CASCADE"后，当删除data表中的某一行时，数据库会自动删除user_data_relation表中所有关联的行。

2. 已创建表后修改外键约束 #

如果表已经存在，无需重建表，可通过执行ALTER TABLE语句修改外键的级联规则：

from sqlalchemy import text

with engine.connect() as conn:
    # 先删除原有外键约束（替换为实际的约束名，不同数据库约束名格式不同）
    conn.execute(text("ALTER TABLE user_data_relation DROP CONSTRAINT user_data_relation_data_id_fkey"))
    # 添加带级联删除的新外键约束
    conn.execute(text("ALTER TABLE user_data_relation ADD CONSTRAINT user_data_relation_data_id_fkey FOREIGN KEY (data_id) REFERENCES data(id) ON DELETE CASCADE"))
    conn.commit()

注意：不同数据库的外键约束名格式有差异，例如MySQL的约束名可能是user_data_relation_ibfk_1，需要先查询数据库获取实际约束名。

验证级联删除 #

完成设置后，执行删除操作即可触发级联：

from sqlalchemy import delete

with engine.connect() as conn:
    # 删除data表中id为1的行，关联的user_data_relation行会被自动删除
    conn.execute(delete(Data).where(Data.c.id == 1))
    conn.commit()

内容的提问来源于stack exchange，提问作者Chris