SQLAlchemy支持递归CTE查询树形结构,需用select().cte(recursive=True)定义锚点与递归成员并union_all合并,注意数据库兼容性、字段类型一致及防无限循环。

SQLAlchemy 支持递归 CTE(Common Table Expression),可用于查询树形结构数据(如组织架构、分类目录、评论回复链等)。核心是使用 select().cte(recursive=True) 构建递归查询,并通过 union_all() 合并锚点(anchor)和递归成员(recursive member)。
基本写法:构造递归 CTE
假设有一张 categories 表,字段为 id、name、parent_id(自引用外键):
from sqlalchemy import select, column, literal_column
from sqlalchemy.orm import Session
<h1>锚点查询:根节点(例如 parent_id is null)</h1><p>anchor = select(
column("id"),
column("name"),
column("parent_id"),
literal_column("0").label("level") # 层级深度
).select_from(table_categories).where(column("parent<em>id").is</em>(None))</p><h1>创建递归 CTE,别名设为 'tree'</h1><p>tree_cte = anchor.cte(name="tree", recursive=True)</p><h1>递归成员:关联上一层的 tree.id == categories.parent_id</h1><p>recursive = select(
column("id"),
column("name"),
column("parent_id"),
(tree_cte.c.level + 1).label("level")
).select_from(table_categories.join(tree_cte, table_categories.c.parent_id == tree_cte.c.id))</p><h1>合并锚点与递归部分</h1><p>cte_query = anchor.union_all(recursive).cte("tree")</p><h1>最终查询:可加排序、过滤等</h1><p>stmt = select(cte_query.c.id, cte_query.c.name, cte_query.c.level).select_from(cte_query).order_by(cte_query.c.level, cte_query.c.id)
使用 ORM 模型时的写法(推荐)
若已定义 ORM 模型(如 Category),需用 aliased 处理自连接,并通过 select(...).select_from(...) 显式指定表源:
from sqlalchemy.orm import aliased
from sqlalchemy import select, literal_column
<p>Category = Base.classes.categories # 或你的实际模型类
cat_alias = aliased(Category) # 用于递归连接的别名</p><h1>锚点:根分类</h1><p>anchor = select(
Category.id,
Category.name,
Category.parent_id,
literal_column("0").label("level")
).where(Category.parent<em>id.is</em>(None))</p><h1>递归 CTE</h1><p>tree = anchor.cte("tree", recursive=True)</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/ai/1552" title="Programming Helper"><img
src="https://img.php.cn/upload/ai_manual/000/000/000/175680263524608.jpg" alt="Programming Helper" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/ai/1552" title="Programming Helper">Programming Helper</a>
<p>AI代码自动生成器,在AI的帮助下更快地编程</p>
</div>
<a href="/ai/1552" title="Programming Helper" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><h1>递归部分:关联到当前树节点的子项</h1><p>tree_alias = aliased(tree)
recursive_part = select(
cat_alias.id,
cat_alias.name,
cat_alias.parent_id,
(tree_alias.c.level + 1).label("level")
).join(
cat_alias, cat_alias.parent_id == tree_alias.c.id
)</p><h1>构建完整 CTE</h1><p>full_tree = anchor.union_all(recursive_part).cte("tree")</p><h1>查询结果(含层级)</h1><p>stmt = select(full_tree.c.id, full_tree.c.name, full_tree.c.level).select_from(full_tree).order_by(full_tree.c.level, full_tree.c.id)</p><h1>执行</h1><p>with Session(engine) as session:
results = session.execute(stmt).all()
关键注意事项
- 数据库支持:PostgreSQL、SQL Server、SQLite 3.8.3+、Oracle、MariaDB 10.2+ 支持递归 CTE;MySQL 8.0+ 也支持,但旧版不支持
-
必须显式命名 CTE:调用
.cte(name="xxx", recursive=True),且name参数不可省略 -
递归字段类型需一致:锚点和递归部分对应列(如
level)的数据类型要兼容,建议都用literal_column或显式 cast -
防止无限循环:确保递归条件最终能终止(如
parent_id不指向自身),生产环境建议加MAXRECURSION(SQL Server)或search depth限制(PostgreSQL 可用SEARCH DEPTH FIRST子句)
扩展:获取某节点的完整路径(从根到当前)
可在 CTE 中维护路径字符串,例如:
anchor = select(
Category.id,
Category.name,
Category.parent_id,
literal_column("ARRAY[CAST(" + str(Category.id) + " AS INTEGER)]").label("path"), # PostgreSQL 数组
literal_column("0").label("level")
).where(Category.parent_id.is_(None))
<h1>递归中拼接 path:tree.path || ARRAY[cat_alias.id]</h1><h1>(具体语法依数据库而定,SQLite/MySQL 可用字符串拼接如 <code>tree.path || '/' || cat_alias.id</code>)</h1><p>这样每行结果就包含从根节点到当前节点的 ID 路径,便于前端渲染面包屑或校验权限。









