如何在SQLAlchemy中统计所有子位置（含间接子节点）的数量？

你的SQLAlchemy代码无法统计间接子节点，核心问题是递归CTE的初始化和关联逻辑错误，没有复刻原生SQL中“针对每个父节点递归遍历所有后代”的逻辑。

错误点 #

• 初始CTE定义错误：Location.query.cte把整个Location表作为递归起始，和原生SQL中“以当前父节点的直接子节点为起始”的逻辑不符。
• 关联逻辑错误：q.c.parent_id == Location.id只能捕获直接父节点关系，无法关联递归出的间接后代。
• 结构未对应子查询：原生SQL是为每个父节点单独嵌套递归CTE统计后代，你的代码没有实现这种一对一的统计逻辑。

正确实现 #

以下代码完全对应原生SQL的逻辑，能统计所有直接+间接子节点数量：

from sqlalchemy import func, select, aliased
from sqlalchemy.orm import Session

def get_location_child_counts(session: Session):
    # 定义递归CTE
    children_cte = select(
        Location.id,
        Location.parent_id
    ).cte(name="children", recursive=True)

    # 递归分支：关联CTE自身，遍历间接子节点
    children_alias = aliased(children_cte, name="children_alias")
    location_alias = aliased(Location, name="location_alias")

    recursive_part = select(
        location_alias.id,
        location_alias.parent_id
    ).join(
        children_alias, children_alias.id == location_alias.parent_id
    )

    # 合并初始与递归部分
    children_cte = children_cte.union_all(recursive_part)

    # 为每个Location.id创建统计子查询
    subquery = select(
        func.count(children_cte.c.id).label("child_count")
    ).filter(children_cte.c.parent_id == Location.id).scalar_subquery()

    # 执行主查询
    result = session.query(
        Location.id,
        subquery.label("child_count")
    ).all()

    return result

关键说明 #

• 递归CTE通过union_all实现深度遍历，覆盖所有间接子节点。
• scalar_subquery()实现原生SQL的嵌套子查询效果，为每个父节点单独统计后代数量。
• 关联条件children_cte.c.parent_id == Location.id确保统计的是当前节点的所有后代。

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