SQLAlchemy自引用层级结构如何获取根节点与叶子节点/根叶配对

1. 查询所有叶子节点 #

叶子节点定义：没有任何子节点的节点，即不存在其他行的parent_id等于当前行的id
直接用NOT EXISTS语法即可实现单句查询，代码如下：

from sqlalchemy import exists
from sqlalchemy.orm import aliased

s = session()
ChildModel = aliased(Model)
# 查询所有叶子节点
leaf_nodes = s.query(Model).filter(
    ~exists().where(ChildModel.parent_id == Model.id)
).all()

2. 批量查询所有根节点的全量后代 #

不需要循环遍历每个根节点单独查询，直接修改递归CTE的起始条件为所有根节点即可，还可以在CTE中新增root_id字段标记每个节点所属的根节点，方便后续统计：

s = session()
# 递归CTE，起始为所有根节点，携带root_id字段
recursive_cte = s.query(
    Model.id,
    Model.parent_id,
    Model.id.label("root_id")  # 根节点的root_id就是自身id
).filter(Model.parent_id.is_(None)).cte(name="recursive_cte", recursive=True)

# 递归关联子节点，继承父节点的root_id
recursive_cte = recursive_cte.union_all(
    s.query(
        Model.id,
        Model.parent_id,
        recursive_cte.c.root_id
    ).filter(Model.parent_id == recursive_cte.c.id)
)

# 一次查询得到所有节点及对应所属根节点id
all_node_with_root = s.query(recursive_cte).all()

3. 查询所有根-叶配对数据 #

在上面递归CTE的基础上，过滤出叶子节点即可得到所有根和对应叶子的配对关系：

from sqlalchemy import exists
from sqlalchemy.orm import aliased

s = session()
ChildModel = aliased(Model)

# 先定义带root_id的递归CTE，和上面逻辑一致
recursive_cte = s.query(
    Model.id.label("node_id"),
    Model.parent_id,
    Model.id.label("root_id")
).filter(Model.parent_id.is_(None)).cte(name="recursive_cte", recursive=True)

recursive_cte = recursive_cte.union_all(
    s.query(
        Model.id.label("node_id"),
        Model.parent_id,
        recursive_cte.c.root_id
    ).filter(Model.parent_id == recursive_cte.c.id)
)

# 过滤出是叶子节点的行，得到根-叶配对
root_leaf_pairs = s.query(
    recursive_cte.c.root_id,
    recursive_cte.c.node_id.label("leaf_id")
).filter(
    ~exists().where(ChildModel.parent_id == recursive_cte.c.node_id)
).all()

如果需要同时拿到根节点和叶子节点的完整Model对象，只需修改查询字段关联对应表即可。

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