sqlalchemy 如何写“递归 CTE”查询树形结构-Python教程-PHP中文网

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

sqlalchemy 如何写“递归 cte”查询树形结构

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
<h1>假设已定义映射类 Department</h1><p>dept = Department.<strong>table</strong></p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill12467" title="OpenClaw 插件安装常见问题排查"><img
														src="https://img.php.cn/upload/skill/001/246/273/6adaf20e695c2374.jpg" alt="OpenClaw 插件安装常见问题排查" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill12467" title="OpenClaw 插件安装常见问题排查" class="overflowclass">OpenClaw 插件安装常见问题排查</a>
													<p class="overflowclass">解决OpenClaw插件安装中常见npm包名错误、权限问题和安装验证的排查方法及最佳实践。</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill12467" title="OpenClaw 插件安装常见问题排查" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><h1>锚点：从 id=1 开始（根部门）</h1><p>anchor = select(dept.c.id, dept.c.name, dept.c.parent_id, literal(0).label('level'))</p><h1>创建递归 CTE，显式声明列名（顺序/类型需与 anchor 一致）</h1><p>cte = anchor.cte(name="tree", recursive=True)
tree = cte.alias("t")</p><h1>递归部分：join 原表找 t.id 的子节点</h1><p>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)</p><h1>合并锚点与递归</h1><p>cte = cte.union_all(recursive)</p><h1>最终查询：选 CTE 中所有字段（可加 order by level 排序）</h1><p>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()
</p>

注意事项与常见问题

数据库支持：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)
<p>cte = cte.union_all(recursive)
stmt = select(cte.c.id, cte.c.name, cte.c.path, cte.c.level)
</p>