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 锚点查询：根节点（例如 parent_id is null）
anchor = select( column("id"), column("name"), column("parent_id"), literal_column("0").label("level") # 层级深度 ).select_from(table_categories).where(column("parentid").is(None))
创建递归 CTE，别名设为 'tree'
tree_cte = anchor.cte(name="tree", recursive=True)
递归成员：关联上一层的 tree.id == categories.parent_id
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))
合并锚点与递归部分
cte_query = anchor.union_all(recursive).cte("tree")
最终查询：可加排序、过滤等
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 Category = Base.classes.categories # 或你的实际模型类 cat_alias = aliased(Category) # 用于递归连接的别名
锚点：根分类
anchor = select( Category.id, Category.name, Category.parent_id, literal_column("0").label("level") ).where(Category.parentid.is(None))
递归 CTE
tree = anchor.cte("tree", recursive=True)
递归部分：关联到当前树节点的子项
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 )
构建完整 CTE
full_tree = anchor.union_all(recursive_part).cte("tree")
查询结果（含层级）
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)
执行
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)) 递归中拼接 path：tree.path || ARRAY[cat_alias.id]
（具体语法依数据库而定，SQLite/MySQL 可用字符串拼接如 tree.path || '/' || cat_alias.id）
这样每行结果就包含从根节点到当前节点的 ID 路径，便于前端渲染面包屑或校验权限。