SQLAlchemy支持递归CTE查询树形结构，需用select().cte(recursive=True)创建并以union_all()合并锚点（根节点）和递归成员（子节点），列名须一致，适用于PostgreSQL、SQLite 3.8.3+等数据库。

SQLAlchemy 支持递归 CTE（Common Table Expression），可用于查询树形结构（如组织架构、评论回复链、分类目录等）。核心是使用 select().cte(recursive=True) 创建递归 CTE，并通过 union_all() 合并锚点查询（根节点）和递归查询（子节点）。

定义递归 CTE 的基本结构

递归 CTE 分两部分：锚点（anchor）和递归成员（recursive member），必须用 union_all() 连接，且递归成员中需引用自身 CTE 名称（通过 cte.c 访问列）。

• 锚点查询返回顶层节点（例如 parent_id is null 或指定 root_id）
• 递归查询通过自连接（JOIN）关联原始表与 CTE，获取下一层子节点
• CTE 必须显式声明列名（推荐用 .columns(...) 或在 select 中命名）

以部门表为例实现树形遍历

假设表结构为：departments(id, name, parent_id)，要查某个部门及其所有子部门（含层级深度）：

from sqlalchemy import select, column, literal
from sqlalchemy.orm import Session
假设已定义映射类 Department
dept = Department.table
锚点：从 id=1 开始（根部门）
anchor = select(dept.c.id, dept.c.name, dept.c.parent_id, literal(0).label('level'))
创建递归 CTE，显式声明列名（顺序/类型需与 anchor 一致）
cte = anchor.cte(name="tree", recursive=True)
tree = cte.alias("t")
递归部分：join 原表找 t.id 的子节点
recursive = select(
dept.c.id,
dept.c.name,
dept.c.parent_id,
(tree.c.level + 1).label('level')
).join(dept, dept.c.parent_id == tree.c.id)
合并锚点与递归
cte = cte.union_all(recursive)
最终查询：选 CTE 中所有字段（可加 order by level 排序）
stmt = select(cte.c.id, cte.c.name, cte.c.parent_id, cte.c.level).order_by(cte.c.level)
results = session.execute(stmt).all()

注意事项与常见问题

• 数据库支持：PostgreSQL、SQL Server

  

  、SQLite 3.8.3+、Oracle、Snowflake 等支持递归 CTE；MySQL 8.0+ 支持，但旧版不支持
• 循环防护：SQLAlchemy 不自动防止无限递归，需确保数据无环，或数据库层面加 MAXRECURSION（SQL Server）或 SEARCH DEPTH FIRST（PostgreSQL）等限制
• 列名一致性：锚点与递归 select 的列数、顺序、类型应一致，否则 CTE 构建失败
• ORM 查询限制：递归 CTE 通常用 Core（select() + 表对象）更直接；若用 ORM，需配合 session.execute() 和 result.scalars() 或手动映射

扩展：带路径路径（path）的树形查询

想生成类似 /IT/Dev/Backend 的路径字段，可在锚点中初始化路径，在递归中拼接：

# 锚点中用 name 初始化 path
anchor = select(
    dept.c.id,
    dept.c.name,
    dept.c.parent_id,
    literal(0).label('level'),
    dept.c.name.label('path')  # 根节点路径即自身名
)
cte = anchor.cte(name="tree", recursive=True)
tree = cte.alias("t")
recursive = select(
    dept.c.id,
    dept.c.name,
    dept.c.parent_id,
    (tree.c.level + 1).label('level'),
    (tree.c.path + '/' + dept.c.name).label('path')  # 拼接路径
).join(dept, dept.c.parent_id == tree.c.id)
cte = cte.union_all(recursive)
stmt = select(cte.c.id, cte.c.name, cte.c.path, cte.c.level)