如何在SQLAlchemy中查询递归实体的完整层级结构

问题背景 #

使用SQLAlchemy 2.0 + aiosqlite存储图结构数据，模型包含SuperNode和Node两个核心类，需要根据ID获取指定SuperNode，并加载其全层级关联数据：

• 自身关联的所有nodes
• 每个Node对应的sub_node及sub_node关联的nodes
• 每个Node的所有层级children及对应sub_node

现有方案中，selectinload仅支持固定层级加载，Python递归+懒加载会因层级过深导致N+1查询，性能极差，因此需要用CTE实现全层级数据的高效加载。

核心思路 #

通过递归CTE一次性抓取所有关联的SuperNode和Node数据，再在Python内存中手动组装ORM关联关系。这种方式仅需少量数据库查询，彻底避免懒加载的性能问题。

具体实现 #

1. 编写递归CTE查询所有关联数据 #

递归CTE分为两部分：

• 锚点查询：获取目标SuperNode及其直接关联的Node
• 递归查询：迭代抓取已获取Node的子节点，以及Node对应的sub_node关联的所有Node，直到无新层级数据

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

# 目标SuperNode的ID
target_supernode_id = 1

# 定义递归CTE
with recursive cte as (
    # 锚点：初始SuperNode及其直接nodes
    select(
        SuperNode.id.label("supernode_id"),
        SuperNode.name.label("supernode_name"),
        Node.id.label("node_id"),
        Node.name.label("node_name"),
        Node.sub_node_id,
        func.null().label("child_node_id"),
        func.null().label("child_node_name"),
    )
    .select_from(SuperNode)
    .join(Node, SuperNode.id == Node.supernode_id)
    .where(SuperNode.id == target_supernode_id)
    
    .union_all(
        # 递归分支1：抓取当前节点的子节点
        select(
            cte.c.supernode_id,
            cte.c.supernode_name,
            cte.c.node_id,
            cte.c.node_name,
            cte.c.sub_node_id,
            Node.id.label("child_node_id"),
            Node.name.label("child_node_name"),
        )
        .select_from(cte)
        .join(association_table, cte.c.node_id == association_table.c.parent_id)
        .join(Node, association_table.c.child_id == Node.id),
        
        # 递归分支2：抓取sub_node对应的所有nodes
        select(
            SuperNode.id.label("supernode_id"),
            SuperNode.name.label("supernode_name"),
            Node.id.label("node_id"),
            Node.name.label("node_name"),
            Node.sub_node_id,
            func.null().label("child_node_id"),
            func.null().label("child_node_name"),
        )
        .select_from(cte)
        .join(SuperNode, cte.c.sub_node_id == SuperNode.id)
        .join(Node, SuperNode.id == Node.supernode_id)
        .where(cte.c.sub_node_id.is_not(None))
    )
):
    # 查询所有涉及的SuperNode（含初始节点和sub_node）
    all_supernodes_stmt = select(SuperNode).where(
        SuperNode.id.in_(
            select(cte.c.supernode_id).union(select(cte.c.sub_node_id))
        )
    )
    # 查询所有涉及的Node（含初始节点、子节点、sub_node关联的节点）
    all_nodes_stmt = select(Node).where(
        Node.id.in_(
            select(cte.c.node_id).union(select(cte.c.child_node_id))
        )
    )

2. 异步执行查询并组装关联结构 #

通过异步会话执行查询，将结果存入字典便于快速查找，再手动组装ORM的关联关系：

async with async_session() as session:
    # 加载所有关联的SuperNode，用ID做键
    supernodes_result = await session.execute(all_supernodes_stmt)
    supernodes_map = {sn.id: sn for sn in supernodes_result.scalars()}
    
    # 加载所有关联的Node，用ID做键
    nodes_result = await session.execute(all_nodes_stmt)
    nodes_map = {n.id: n for n in nodes_result.scalars()}
    
    # 组装Node的children关联
    child_relations = await session.execute(
        select(association_table.c.parent_id, association_table.c.child_id)
        .where(association_table.c.parent_id.in_(nodes_map.keys()))
    )
    for parent_id, child_id in child_relations:
        nodes_map[parent_id].children.append(nodes_map[child_id])
    
    # 组装Node的sub_node关联
    for node in nodes_map.values():
        if node.sub_node_id:
            node.sub_node = supernodes_map[node.sub_node_id]
    
    # 组装目标SuperNode的nodes关联
    target_supernode = supernodes_map[target_supernode_id]
    target_supernode.nodes = [
        n for n in nodes_map.values() if n.supernode_id == target_supernode_id
    ]
    
    # 此时target_supernode已包含全层级关联数据
    print(target_supernode.name)
    for node in target_supernode.nodes:
        print(f"  Node: {node.name}, SubNode: {node.sub_node.name if node.sub_node else None}")
        for child in node.children:
            print(f"    Child: {child.name}")

原理与优势 #

1. 递归CTE的作用：一次性遍历所有层级的关联数据，将原本需要N+1次的懒加载查询合并为3次批量查询（SuperNode、Node、父子关系），大幅降低数据库交互开销。
2. 内存组装关联：SQLAlchemy的ORM无法直接映射递归CTE的多层关联，因此先将所有数据加载到内存，通过字典快速查找并手动组装关联关系，避免ORM自动懒加载的触发。
3. 对比其他方案：
   • selectinload：仅支持指定固定层级（如selectinload(SuperNode.nodes).selectinload(Node.children)），无法处理无限层级。
   • Python递归+懒加载：每访问一个未加载的关联都会触发新查询，层级越深查询次数越多，性能呈指数下降。

注意事项 #

• 去重处理：递归CTE可能返回重复数据，通过union和字典去重，确保每个SuperNode和Node仅加载一次。
• 异步适配：使用aiosqlite时，所有查询必须通过await session.execute()执行，不能使用同步方法。
• 性能边界：若数据量极大，可添加递归深度限制或分页逻辑，避免单次查询返回过多数据。

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