在使用sqlalchemy进行sqlite数据库操作时，当通过`session.query(...).delete()`执行批量删除并期望外键的`on delete cascade`行为生效时，可能会发现子记录并未被级联删除。这是因为sqlite默认禁用外键约束，且sqlalchemy的批量删除操作直接发送sql语句，绕过了orm层面的级联逻辑。本文将详细阐述此问题的原因，并提供通过事件监听器在每次连接时开启`pragma foreign_keys=on`的解决方案，确保数据库层面的级联删除正确执行。
深入理解SQLAlchemy与SQLite级联删除问题
在使用SQLAlchemy与SQLite数据库时，开发者可能会遇到一个常见的问题：即使在模型中正确配置了ForeignKey的ondelete="CASCADE"属性，并通过relationship设置了cascade="all, delete"和passive_deletes=True，当使用session.query(Model).filter_by(...).delete()方法删除父记录时，相关的子记录却未能被级联删除。然而，如果直接在SQLite工具中执行相同的SQL DELETE语句，级联删除却能正常工作。
这个问题的核心在于两个方面：
- SQLite的外键约束默认行为： SQLite数据库为了兼容性或性能考量，默认情况下不强制执行外键约束。这意味着即使表结构中包含了ON DELETE CASCADE定义，除非显式启用，否则数据库并不会自动执行级联操作。
- SQLAlchemy的批量删除机制： 当使用session.query(Model).filter_by(...).delete()时，SQLAlchemy会生成并直接执行一条DELETE SQL语句到数据库。这种操作是“批量”且“被动”的，它绕过了SQLAlchemy ORM层面的级联处理逻辑。ORM的cascade设置（如cascade="all, delete"）主要在通过session.delete(instance)删除单个已加载的ORM对象时生效。因此，对于query(...).delete()这种直接的SQL操作，级联行为完全依赖于数据库本身的外键约束。
由于SQLite默认不启用外键约束，当SQLAlchemy发出直接的DELETE语句时，数据库不会触发ON DELETE CASCADE，从而导致子记录未能被删除。
解决方案：在SQLite连接时启用外键约束
要解决此问题，我们需要确保每次SQLAlchemy建立与SQLite数据库的连接时，都显式地开启外键约束。这可以通过SQLAlchemy的事件系统来实现，在每次数据库连接建立时执行PRAGMA foreign_keys=ON语句。
以下是实现此功能的代码示例：
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()将这段代码添加到你的SQLAlchemy应用初始化部分，通常是在定义Engine之后，确保在创建任何会话之前被执行。
完整示例代码
为了更好地理解和应用上述解决方案，我们来看一个完整的示例，包括模型定义、引擎创建、事件监听器设置以及删除操作：
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()}")注意事项与总结
- PRAGMA foreign_keys=ON的重要性： 这是解决SQLite级联删除问题的关键。没有它，即使数据库表结构定义了ON DELETE CASCADE，SQLite也不会执行。
-
session.query(...).delete()与session.delete(instance)的区别：
- session.query(...).delete()：直接向数据库发送SQL DELETE语句，效率高，但不触发ORM层面的级联。级联行为完全依赖于数据库的外键约束。
- session.delete(instance)：删除一个已加载到会话中的ORM对象。这会触发ORM层面的cascade设置（如cascade="all, delete"），SQLAlchemy会负责加载并删除相关联的子对象。
- passive_deletes=True的作用： 当你在relationship中同时设置了passive_deletes=True和ondelete="CASCADE"（在外键上）时，它告诉SQLAlchemy，当父对象被删除时，ORM不需要去加载并删除子对象，而是可以信赖数据库的ON DELETE CASCADE机制来完成。这是一种性能优化，避免了不必要的子对象加载。
- 验证DDL： 始终检查SQLAlchemy生成的CREATE TABLE语句，确保FOREIGN KEY定义中包含了ON DELETE CASCADE。
通过在每次SQLite连接时启用PRAGMA foreign_keys=ON，我们可以确保SQLAlchemy在执行session.query(...).delete()等批量操作时，数据库能够正确地执行外键级联删除，从而保持数据的一致性。
