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)
