SQLAlchemy无关联关系下一对多连接结果结构化问题

方法一：Python代码层面聚合结果 #

先执行原查询拿到平铺数据，再手动整理成嵌套结构：

# 执行原始查询获取所有结果
results = db.query(table_1, table_2).join(table_2, table_2.object_id == table_1.id).all()

# 初始化字典用于聚合数据
aggregated_data = {}
for t1_record, t2_record in results:
    t1_id = t1_record.id
    if t1_id not in aggregated_data:
        aggregated_data[t1_id] = {
            "table_1_id": t1_id,
            "table_2_data": []
        }
    aggregated_data[t1_id]["table_2_data"].append({"data": t2_record.data})

# 转换为期望的列表格式
final_result = list(aggregated_data.values())

这种方式不依赖表的关系定义，逻辑简单直接，适合数据量不大的场景。

方法二：用PostgreSQL聚合函数直接生成嵌套结构 #

利用PostgreSQL的json_agg函数在SQL层面完成聚合，直接返回嵌套格式：

from sqlalchemy import func

# 构造聚合查询
query = db.query(
    table_1.id.label("table_1_id"),
    func.json_agg(func.json_build_object("data", table_2.data)).label("table_2_data")
).join(table_2, table_2.object_id == table_1.id).group_by(table_1.id)

# 转换为字典格式结果
final_result = [row._asdict() for row in query.all()]

该方法将聚合逻辑交给数据库处理，效率更高，适合大数据量场景，核心是通过json_agg把多条table_2记录打包成JSON数组。

之前方法失效的原因 #

• group_by：单独使用仅能分组，必须配合聚合函数（如json_agg）才能生成嵌套数组，之前未正确搭配使用；
• subqueryload：属于SQLAlchemy的ORM关系加载特性，需要在模型类中提前定义relationship关联，但pipelinewise-target-postgres创建的表未生成对应ORM关系，因此无法生效；
• distinct：仅用于去重，不具备数据聚合嵌套的能力。

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