SQLAlchemy递归查询实现求助：获取容器及子容器所有项目

我懂你这种卡在递归查询上的感觉——嵌套容器+关联项目的需求，用SQLAlchemy确实容易找不到精准的方案。我给你整理了两种实用的解决思路，适配不同场景：

先明确表模型（对应你的table1和table2） #

首先咱们把表结构用SQLAlchemy的ORM模型定义清楚，方便后续代码对应：

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

Base = declarative_base()

# 对应你的table1：容器表，支持嵌套（子容器关联自身）
class Container(Base):
    __tablename__ = 'table1'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # 子容器的父容器ID，可为空表示顶级容器
    parent_container_id = Column(Integer, ForeignKey('table1.id'), nullable=True)
    
    # ORM关联关系：父容器
    parent = relationship('Container', remote_side=[id], back_populates='children')
    # ORM关联关系：子容器集合
    children = relationship('Container', back_populates='parent')
    # ORM关联关系：当前容器下的所有项目
    items = relationship('Item', back_populates='container')

# 对应你的table2：项目表，关联到容器
class Item(Base):
    __tablename__ = 'table2'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    container_id = Column(Integer, ForeignKey('table1.id'), nullable=False)
    
    container = relationship('Container', back_populates='items')

方案1：数据库层面递归CTE（推荐，高效处理大数据） #

这种方式利用数据库的递归CTE（Common Table Expression）能力，直接在数据库层面完成递归查询，效率远高于Python层面的递归，适合数据量较大的场景：

from sqlalchemy import select, union_all

def get_all_items_for_container(container_id, session):
    # 第一步：定义递归CTE的初始部分——获取目标容器的直接关联项目
    cte = select(
        Container.id.label('container_id'),
        Item.id.label('item_id'),
        Item.name.label('item_name')
    ).join(Item, Container.id == Item.container_id)\
     .where(Container.id == container_id)\
     .cte(recursive=True)
    
    # 第二步：定义递归部分——获取所有子容器的项目，关联到CTE中的父容器ID
    recursive_part = select(
        Container.id.label('container_id'),
        Item.id.label('item_id'),
        Item.name.label('item_name')
    ).join(Item, Container.id == Item.container_id)\
     .join(cte, Container.parent_container_id == cte.c.container_id)
    
    # 第三步：合并初始部分和递归部分，形成完整的递归查询
    full_cte = cte.union_all(recursive_part)
    
    # 执行查询并整理结果
    result = session.execute(full_cte).fetchall()
    # 这里返回字典格式，你也可以根据需求转换成Item对象
    return [{'item_id': row.item_id, 'item_name': row.item_name} for row in result]

调用方式很简单：

# 假设你已经创建了SQLAlchemy的session
all_items = get_all_items_for_container(target_container_id, session)

方案2：ORM层面递归（适合小数据量/测试场景） #

如果你的数据量不大，或者更倾向于用Python代码逻辑来控制递归，可以直接通过ORM的关系遍历嵌套容器，收集所有项目：

def get_all_items_recursively(container):
    items = []
    # 添加当前容器的所有项目
    items.extend(container.items)
    # 递归遍历每个子容器，收集子容器的项目
    for child_container in container.children:
        items.extend(get_all_items_recursively(child_container))
    return items

# 使用方式：先获取目标容器，再调用递归函数
target_container = session.query(Container).get(target_container_id)
all_items = get_all_items_recursively(target_container)

优化提示：避免N+1查询 #

如果用这种方式，记得提前加载关联的子容器和项目，避免频繁查询数据库：

target_container = session.query(Container)\
    .options(
        joinedload(Container.children).joinedload(Container.items),
        joinedload(Container.items)
    )\
    .get(target_container_id)

注意事项 #

• 递归CTE需要数据库支持：比如MySQL 8.0+、PostgreSQL、SQL Server等都支持，如果你用的是老版本MySQL（比如5.x），那只能用方案2或者手动实现递归逻辑。
• 如果需要返回完整的容器层级结构，还可以在CTE中添加层级字段，方便后续区分项目属于哪个层级的容器。

内容的提问来源于stack exchange，提问作者Víctor López