SQLAlchemy+SQLite外键级联删除失效：关联表行未被删除

使用SQLAlchemy搭配SQLite（aiosqlite驱动），已配置外键级联规则与ORM关联的cascade属性，但执行Accounts记录删除操作时，仅accounts表的行被删除，关联的emails表对应行无变化。

相关代码片段 #

数据表定义 #

class AccountsModel(Base):
    __tablename__ = 'accounts'

    id: Mapped[intpk]
    email_id: Mapped[int] = mapped_column(
        ForeignKey('emails.id', ondelete='CASCADE')
    )
    new_email_id: Mapped[int | None] = mapped_column(
        ForeignKey('new_emails.id', ondelete='CASCADE'),
        nullable=True
    )
    email: Mapped['EmailsModel'] = relationship(
        back_populates='account',
        single_parent=True,
        cascade='all, delete-orphan',
        lazy='joined'
    )
    new_email: Mapped['NewEmailsModel'] = relationship(
        back_populates='account',
        single_parent=True,
        cascade='all, delete-orphan',
        lazy='joined'
    )


class EmailsModel(Base):
    __tablename__ = 'emails'

    id: Mapped[intpk]
    email: Mapped[str_255] = mapped_column(unique=True)
    account: Mapped['AccountsModel'] = relationship(
        back_populates='email',
        cascade='all, delete-orphan'
    )

引擎配置 #

class SessionMaker:
    def __init__(self, dsn: str):
        self._engine = create_async_engine(dsn)

    @staticmethod
    @event.listens_for(Engine, 'connect')
    def _set_sqlite_pragma(connection: Any, record: '_ConnectionRecord'):
        cursor = connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON')
        cursor.close()

    async def create_tables(self, drop=False) -> None:
        async with self._engine.begin() as conn:
            if drop:
                await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)

    @property
    def session(self) -> async_sessionmaker[AsyncSession]:
        if not hasattr(self, '_session'):
            self._session = async_sessionmaker(self._engine)
        return self._session

删除逻辑 #

async def delete(
    self,
    *,
    filter_by: Dict[str, Any]
) -> None:
    async with self._session() as session:
        stmt = (
            delete(self._model)
            .filter_by(**filter_by)
        )
        await session.execute(stmt)
        await session.commit()

async def delete_by_id(
    self,
    *,
    id: int,
) -> None:
    await self.delete(filter_by=dict(id=id))

await MY_MODEL.delete_by_id(MY_ID)

现象 #

执行删除后，日志仅显示accounts表的删除语句，emails表无操作：

BEGIN (implicit)
DELETE FROM accounts WHERE accounts.id = ?
[cached since 20.71s ago] (3,)
COMMIT

表创建语句已正确生成外键规则：

CREATE TABLE accounts (
    id INTEGER NOT NULL,
    email_id INTEGER NOT NULL,
    new_email_id INTEGER,
    PRIMARY KEY (id),
    FOREIGN KEY(email_id) REFERENCES emails (id) ON DELETE CASCADE,
    FOREIGN KEY(new_email_id) REFERENCES new_emails (id) ON DELETE CASCADE
)
CREATE TABLE emails (
    id INTEGER NOT NULL,
    email VARCHAR(255) NOT NULL,
    PRIMARY KEY (id),
    UNIQUE (email)
)

问题排查与解决方案 #

1. 外键级联方向错误 #

当前accounts表的email_id外键设置ON DELETE CASCADE，作用是：当emails表的被引用行删除时，自动删除accounts表中关联的行，和你想要的“删除accounts行时级联删emails行”方向完全相反。

修复方式（数据库级联）

调整外键到emails表，让emails依赖accounts：

# 修改EmailsModel
class EmailsModel(Base):
    __tablename__ = 'emails'

    id: Mapped[intpk]
    email: Mapped[str_255] = mapped_column(unique=True)
    # 添加指向accounts的外键并设置级联
    account_id: Mapped[int] = mapped_column(ForeignKey('accounts.id', ondelete='CASCADE'))
    account: Mapped['AccountsModel'] = relationship(
        back_populates='email',
        cascade='all, delete-orphan'
    )

# 修改AccountsModel，移除原email_id外键
class AccountsModel(Base):
    __tablename__ = 'accounts'

    id: Mapped[intpk]
    new_email_id: Mapped[int | None] = mapped_column(
        ForeignKey('new_emails.id', ondelete='CASCADE'),
        nullable=True
    )
    email: Mapped['EmailsModel'] = relationship(
        back_populates='account',
        single_parent=True,
        cascade='all, delete-orphan',
        lazy='joined'
    )
    new_email: Mapped['NewEmailsModel'] = relationship(
        back_populates='account',
        single_parent=True,
        cascade='all, delete-orphan',
        lazy='joined'
    )

重新生成表后，删除accounts行时，数据库会自动删除关联的emails行。

2. 批量DELETE语句不触发ORM级联 #

你当前使用的delete(self._model).filter_by(...)是直接生成SQL执行的批量删除操作，不会触发ORM层面的cascade规则——ORM的cascade仅在调用session.delete(obj)删除加载到内存的对象时生效。

修复方式（ORM级联）

修改删除逻辑，先加载目标对象再删除：

async def delete_by_id(
    self,
    *,
    id: int,
) -> None:
    async with self._session() as session:
        obj = await session.get(self._model, id)
        if obj:
            session.delete(obj)
            await session.commit()

这种方式会触发ORM的cascade='all, delete-orphan'规则，删除accounts对象的同时删除关联的email对象。

3. 确认SQLite外键功能已启用 #

虽然你配置了PRAGMA foreign_keys=ON，但异步引擎的connect事件可能未正确触发，可手动验证：

async def check_foreign_keys_status(self):
    async with self._engine.connect() as conn:
        result = await conn.execute(text("PRAGMA foreign_keys"))
        print("SQLite外键启用状态：", result.scalar())

返回1表示已启用，0则需检查事件监听逻辑是否生效。

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