我用SQLAlchemy Core定义了一个自关联的分类表，结构如下：

categories = Table("categories", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("parent_id", Integer, ForeignKey("categories.id")),
    Column("dept_id", Integer, ForeignKey("departments.id")),
    CheckConstraint('id!=parent_id'),
    nullable=True),
    UniqueConstraint('parent_id','name', name='parent_category')

我期望生成嵌套层级的JSON格式，示例如下：

{"id": 1, "parent_id": null, "name": "root", "dept_id": null, "children":[ 
    {"id": 2, "parent_id": 1, "name": "Top_level", "dept_id": null}, 
    {"id": 3, "parent_id": 1, "name": "Rt_Docs", "dept_id": null} 
]}

我尝试用嵌套CTE的SQL语句来实现，但输出的结果不符合预期——每个层级都会重复输出完整树结构，比如：

"cat_list": [ 
    { "jsonb_agg": [ 
        { "Name": "root", "Sub-cat": [ 
            { "Name": "dne-mount" }, 
            { "Name": "dne-mktg" }, 
            { "Name": "Marketing" }, 
            { "Name": "Top_level" }, 
            { "Name": "Rt_Docs" } 
        ] }, 
        { "Name": "root", "Sub-cat": { "Name": "Top_level", "Sub-cat": [ { "Name": "Cat1" } ] }
    ]}
]

我用的SQLAlchemy代码如下：

s = text('with recursive rec_cat as(select id, parent_id, name, dept_id, 0 as level, array[id] as path_info from categories where parent_id is null union all select cat1.id, cat1.parent_id, cat1.name, cat1.dept_id, level + 1, path_info||cat1.parent_id from rec_cat p join categories cat1 on cat1.parent_id=p.id where not cat1.id= any(path_info)), dncc as (select cat1.parent_id, json_agg(jsonb_build_object(\'Name\',cat1.name))::jsonb as js from rec_cat tree join categories cat1 using (id) where level > 0 and not id=any(path_info) group by cat1.parent_id union all select cat1.parent_id, jsonb_build_object(\'Name\', cat1.name) || jsonb_build_object(\'Sub-cat\', js) as js from dncc tree join categories cat1 on cat1.id=tree.parent_id) select jsonb_agg(js) from dncc where parent_id is null')
cat_list = []
mdept_id = None
rs = g.conn.execute(s)
for r in rs:
    mx = collections.OrderedDict(r)
    cat_list.append(mx)
return jsonify({'cat_list': cat_list}), 200

需要指导修改哪些部分，生成符合预期的嵌套JSON格式。

问题出在你的递归CTE构建JSON的逻辑上，当前的dncc部分没有正确地从叶子节点向上聚合嵌套结构，而是重复生成了不同层级的完整树。我们可以调整递归CTE的逻辑，从叶子节点开始向上构建包含children的嵌套对象，或者直接在递归中构建完整的嵌套结构。

方法1：调整递归CTE，正确向上聚合嵌套JSON #

我们可以重新设计递归CTE，让它从最底层的节点开始，逐步向上将子节点聚合到父节点的children字段中：

s = text('''
WITH RECURSIVE rec_cat AS (
    -- 基础查询：获取所有节点，标记是否为叶子节点
    SELECT 
        id, parent_id, name, dept_id,
        NOT EXISTS (SELECT 1 FROM categories c WHERE c.parent_id = categories.id) AS is_leaf
    FROM categories
),
-- 递归部分：从叶子节点开始，向上构建嵌套结构
nested_cats AS (
    -- 初始叶子节点：没有子节点，所以children为空数组
    SELECT 
        id, parent_id,
        jsonb_build_object(
            'id', id,
            'parent_id', parent_id,
            'name', name,
            'dept_id', dept_id,
            'children', '[]'::jsonb
        ) AS node
    FROM rec_cat
    WHERE is_leaf = TRUE

    UNION ALL

    -- 递归步骤：将子节点的node聚合到父节点的children中
    SELECT 
        p.id, p.parent_id,
        jsonb_build_object(
            'id', p.id,
            'parent_id', p.parent_id,
            'name', p.name,
            'dept_id', p.dept_id,
            'children', jsonb_agg(c.node)
        ) AS node
    FROM rec_cat p
    JOIN nested_cats c ON p.id = c.parent_id
    WHERE p.is_leaf = FALSE
    GROUP BY p.id, p.parent_id, p.name, p.dept_id
)
-- 最终获取根节点（parent_id为null）的嵌套结构
SELECT jsonb_agg(node) AS cat_list
FROM nested_cats
WHERE parent_id IS NULL
''')

cat_list = []
rs = g.conn.execute(s)
for r in rs:
    cat_list.extend(r['cat_list'])  # 直接提取聚合后的数组
return jsonify({'cat_list': cat_list}), 200

逻辑说明： #

1. rec_cat CTE：先获取所有节点，并标记哪些是叶子节点（没有子节点的节点）。
2. nested_cats CTE：
   • 初始部分：处理叶子节点，生成包含基本字段和空children数组的JSON对象。
   • 递归部分：将子节点的JSON对象聚合到父节点的children字段中，逐步向上构建完整的嵌套结构。
3. 最后查询根节点（parent_id IS NULL）的结果，就是我们需要的嵌套JSON。

方法2：用SQLAlchemy的递归查询API替代原生SQL（可选） #

如果你不想写原生SQL，也可以用SQLAlchemy的递归查询构造器来实现，代码可读性更好：

from sqlalchemy import select, func, exists, union_all

# 基础CTE：获取所有节点和叶子标记
rec_cat = select(
    categories.c.id,
    categories.c.parent_id,
    categories.c.name,
    categories.c.dept_id,
    ~exists().where(categories.c.id == categories.c.parent_id).label("is_leaf")
).cte(recursive=True)

# 递归部分的初始查询（叶子节点）
leaf_nodes = select(
    rec_cat.c.id,
    rec_cat.c.parent_id,
    func.jsonb_build_object(
        'id', rec_cat.c.id,
        'parent_id', rec_cat.c.parent_id,
        'name', rec_cat.c.name,
        'dept_id', rec_cat.c.dept_id,
        'children', '[]'::func.jsonb
    ).label('node')
).where(rec_cat.c.is_leaf == True)

# 递归部分的父节点查询
parent_nodes = select(
    rec_cat.c.id,
    rec_cat.c.parent_id,
    func.jsonb_build_object(
        'id', rec_cat.c.id,
        'parent_id', rec_cat.c.parent_id,
        'name', rec_cat.c.name,
        'dept_id', rec_cat.c.dept_id,
        'children', func.jsonb_agg(leaf_nodes.c.node)
    ).label('node')
).join(leaf_nodes, rec_cat.c.id == leaf_nodes.c.parent_id
).where(rec_cat.c.is_leaf == False
).group_by(rec_cat.c.id, rec_cat.c.parent_id, rec_cat.c.name, rec_cat.c.dept_id)

# 组合递归CTE
nested_cats = rec_cat.union_all(leaf_nodes, parent_nodes)

# 最终查询根节点
final_query = select(func.jsonb_agg(nested_cats.c.node).label('cat_list')).where(nested_cats.c.parent_id == None)

rs = g.conn.execute(final_query)
result = rs.scalar()  # 获取聚合后的JSON数组
return jsonify({'cat_list': result}), 200

关键修改点总结 #

• 你的原CTE在dncc部分没有正确区分层级聚合，导致重复生成树结构；新的逻辑从叶子节点向上聚合，确保每个父节点只聚合它的直接子节点。
• 改用jsonb_build_object生成包含所有字段（id、parent_id、name、dept_id、children）的对象，而不是只生成Name和Sub-cat，贴合你的期望输出格式。
• 最终只提取根节点的结果，避免重复输出。

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