如何在SQLAlchemy中删除所有无Child的Parent对象？

参考模型 #

首先参考SQLAlchemy的基础关系模型：

class Parent(Base):
    __tablename__ = "parent_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    children: Mapped[List["Child"]] = relationship(back_populates="parent")


class Child(Base):
    __tablename__ = "child_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent_table.id"))
    parent: Mapped["Parent"] = relationship(back_populates="children")

问题 #

如何删除所有没有关联Child对象的Parent？

尝试的错误代码 #

from sqlalchemy import func
from sqlalchemy.orm import Session

with Session(engine) as sesssion:
    session.query(Parent).filter(
        func.count(Parent.children) == 0
    ).execution_options(is_delete_using=True).delete()

错误信息 #

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) 聚合函数count()使用不当
[SQL: SELECT parent_table.id
FROM parent_table, child_table
WHERE count(parent_table.id = child_table.parent_id) = ?]
[parameters: (0,)]

错误原因 #

直接在WHERE子句中使用聚合函数count()不符合SQL语法规范，聚合函数通常需要配合GROUP BY子句使用，或通过子查询、EXISTS/NOT EXISTS判断关联关系。SQLAlchemy根据错误写法生成的SQL存在逻辑问题，导致SQLite报错。

正确解决方案 #

方案一：使用NOT EXISTS子查询（推荐，高效） #

通过子查询判断是否不存在关联的Child记录，适合大数据量场景：

from sqlalchemy import exists
from sqlalchemy.orm import Session

with Session(engine) as session:
    # 构造子查询：判断当前Parent是否存在关联的Child
    has_child = session.query(Child).filter(Child.parent_id == Parent.id).exists()
    # 筛选出没有关联Child的Parent并删除
    session.query(Parent).filter(~has_child).delete(synchronize_session=False)
    session.commit()

delete()方法中synchronize_session=False表示不需要同步ORM会话中的对象，能提升批量删除的效率。

方案二：使用LEFT JOIN + IS NULL #

通过左连接筛选出未匹配到Child的Parent：

from sqlalchemy.orm import Session

with Session(engine) as session:
    parents_to_delete = session.query(Parent).\
        outerjoin(Child, Parent.id == Child.parent_id).\
        filter(Child.id.is_(None))
    parents_to_delete.delete(synchronize_session=False)
    session.commit()

方案三：ORM层面筛选（适合小数据量） #

利用ORM的relationship直接判断子集合是否为空，这种方式会先将Parent对象加载到内存中，仅适合数据量较小的场景：

from sqlalchemy.orm import Session

with Session(engine) as session:
    # 筛选出children为空的Parent
    parents = session.query(Parent).filter(Parent.children == []).all()
    for parent in parents:
        session.delete(parent)
    session.commit()

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