如何在SQLAlchemy递归查询中按原始子分类ID分组结果？

当然可以实现！我们有两种思路来满足你的需求：一种是修改CTE递归逻辑，在结果中保留原始子分类ID（方便后续分组，性能更优），另一种是为每个子分类单独执行递归查询（直接得到分组结果，简单直观）。下面分别详细说明：

方法一：修改CTE，保留原始子分类ID（推荐） #

核心思路是在递归的初始查询中，为每个子分类标记它的「原始ID」（即id_list中的目标ID），然后在递归查询父节点时，把这个原始ID传递下去。最终结果会包含每个节点对应的原始子分类ID，你可以轻松用Python完成分组。

修改后的代码实现 #

# 假设fields是你要查询的字段列表，比如[Category.name]
r = session.query(
    Category.id,
    Category.parent_id,
    *fields,
    Category.id.label('original_id')  # 初始时，原始ID就是子分类自身ID
).filter(Category.id.in_(id_list)).cte(name='r', recursive=True)

r_alias = aliased(r, name="recursive")
base_alias = aliased(Category, name='base')

# 递归查询父节点时，传递原始ID
included_parts = r.union_all(
    session.query(
        base_alias.id,
        base_alias.parent_id,
        *[getattr(base_alias, field.key) if isinstance(field, InstrumentedAttribute) else field for field in fields],
        r_alias.c.original_id  # 保留原始子分类ID
    ).filter(base_alias.id == r_alias.c.parent_id)
)

# 执行查询
q = session.query(included_parts).all()
print(q)

查询结果示例 #

你会得到带原始ID的结果（original_id为元组最后一个元素）：

[(1, None, 'Category1', 1), (2, 1, 'Category2', 2), (1, None, 'Category1', 2)]

用Python分组得到目标格式 #

使用itertools.groupby按original_id分组：

from itertools import groupby
from operator import itemgetter

# groupby要求连续相同键，所以先按original_id排序
sorted_results = sorted(q, key=itemgetter(-1))
grouped_result = [list(group) for _, group in groupby(sorted_results, key=itemgetter(-1))]
print(grouped_result)

最终分组结果 #

[[(1, None, 'Category1', 1)], [(2, 1, 'Category2', 2), (1, None, 'Category1', 2)]]

方法二：为每个子分类单独执行递归查询（简单直接） #

如果id_list的规模不大，你可以循环遍历每个子分类ID，单独执行递归查询，直接得到分组后的结果：

代码示例 #

grouped_result = []
for cat_id in id_list:
    # 为当前子分类单独构建CTE递归查询
    r = session.query(Category.id, Category.parent_id, *fields).\
        filter(Category.id == cat_id).\
        cte(name='r', recursive=True)
    r_alias = aliased(r)
    base_alias = aliased(Category)
    included_parts = r.union_all(
        session.query(base_alias).filter(base_alias.id == r_alias.c.parent_id)
    )
    # 查询当前子分类的所有父节点（包括自身）
    sub_result = session.query(included_parts).all()
    grouped_result.append(sub_result)

print(grouped_result)

结果示例 #

直接得到你期望的格式：

[[(1, None, 'Category1')], [(2, 1, 'Category2'), (1, None, 'Category1')]]

注意事项 #

这种方法的缺点是如果id_list包含大量ID，会执行多次数据库查询，性能不如方法一。如果数据量较大，优先选择方法一。

内容的提问来源于stack exchange，提问作者Bob