SQLAlchemy中如何递归joinload未知深度的树形对象？

当然可以，核心思路是利用**递归CTE（Common Table Expression）**来一次性查询所有层级的节点及其关联数据——ORM自带的joinedload/selectinload只能处理固定层级的关联，没法应对不确定深度的树形结构。

具体实现步骤： #

1. 编写递归CTE查询

递归CTE分为锚点查询（获取根节点）和递归查询（关联子节点）两部分，同时在查询中直接join需要的关联表，一次性拉取所有关联数据。

假设你的模型结构如下：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class Config(Base):
    __tablename__ = 'configs'
    id = Column(Integer, primary_key=True)
    value = Column(String)

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    parent_id = Column(Integer, ForeignKey('categories.id'))
    config_id = Column(Integer, ForeignKey('configs.id'))
    
    children = relationship('Category', backref='parent')
    config = relationship('Config')

对应的递归CTE查询代码：

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

def load_full_tree(session: Session):
    # 锚点查询：获取所有根节点（parent_id为None）并关联Config
    cte = select(
        Category.id,
        Category.name,
        Category.parent_id,
        Config.id.label('config_id'),
        Config.value.label('config_value')
    ).join(Config, Category.config_id == Config.id)\
     .where(Category.parent_id.is_(None))\
     .cte(recursive=True)
    
    # 递归部分：关联子节点及其Config
    child_cat = aliased(Category)
    child_config = aliased(Config)
    cte = cte.union_all(
        select(
            child_cat.id,
            child_cat.name,
            child_cat.parent_id,
            child_config.id.label('config_id'),
            child_config.value.label('config_value')
        ).join(child_config, child_cat.config_id == child_config.id)\
         .join(cte, cte.c.id == child_cat.parent_id)
    )
    
    # 执行查询
    return session.execute(cte).fetchall()

2. 将扁平结果组装为树形结构

CTE返回的是扁平的结果集，需要手动转换成嵌套的树形结构，方便渲染使用：

def build_tree(result_set):
    # 先把所有节点存入字典，方便快速查找
    node_map = {}
    for row in result_set:
        node_map[row.id] = {
            'id': row.id,
            'name': row.name,
            'config': {'id': row.config_id, 'value': row.config_value},
            'children': []
        }
    
    # 构建树形结构
    tree_root = []
    for node in node_map.values():
        if node['parent_id'] is None:
            tree_root.append(node)
        else:
            parent_node = node_map.get(node['parent_id'])
            if parent_node:
                parent_node['children'].append(node)
    return tree_root

3. 调用方式

# 假设session是已创建的SQLAlchemy会话
flat_result = load_full_tree(session)
tree = build_tree(flat_result)
# 此时tree就是包含所有层级节点及关联Config的树形结构，可直接用于渲染

注意事项： #

• 递归深度限制：不同数据库对递归CTE的深度有默认限制（比如PostgreSQL默认是1000层），如果你的树形结构超过这个限制，需要调整数据库配置。
• 性能优化：如果节点数量极大，建议在CTE中添加过滤条件，只查询需要的分支，避免一次性拉取过多数据。
• 多关联场景：如果模型有多个关联表，只需在锚点和递归查询中分别join对应的表，把需要的字段加入select语句即可。

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