Python如何将数据库返回的扁平化一对多记录转换为嵌套结构化对象/JSON

假设存在学生→书籍的一对多关联关系，对应关系示意图如下：

使用Python查询关联数据表时，经常会得到扁平化的返回结果，例如使用psycopg执行原生SQL联表查询，会得到如下格式的记录：

records = [
    (1, "Student 1", 11, "Book 1"),
    (1, "Student 1", 12, "Book 2"),
    (1, "Student 1", 13, "Book 3"),
    (2, "Student 2", 21, "Book 4"),
    (2, "Student 2", 22, "Book 5"),
    ...
]

联表查询会导致父级记录（如学生信息）在多行中重复出现，实际业务中关联关系往往不止两层，常见的多层关联如学生→书籍→作者→所属公司，需要将这类扁平化结果组装为符合业务逻辑的嵌套结构，目标格式示例如下：

[
   {
      "id": 1,
      "name": "Student 1",
      "books": [
         {
           "id": 11,
           "name": "Book 1",
           "author": {            
             "id": 100,
             "name": "John Doe",
             "company": {
                "id": "...",
                "addresses": [
                    {...},
                    {...}
                ] 
             }
           }
         },
         {
           "id": 12,
           "name": "Book 12"
         },
         {
           "id": 13,
           "name": "Book 13"
         }
      ]
   }
]

Java生态中jOOQ内置了该类关联结果的自动组装能力，Python生态可通过以下几类方案实现同等效果。

可用方案 #

• 主流ORM内置能力
  如果项目本身已经在使用ORM框架，不需要额外引入工具，框架原生就支持关联结果的嵌套组装：

  • SQLAlchemy：使用joinedload、selectinload做关联预加载时，返回的模型对象本身就是嵌套结构，配合Pydantic、marshmallow等序列化工具即可直接输出目标JSON格式，支持任意层级的一对多、多对一、多对多关系，自动处理重复父记录的去重。
  • Peewee：轻量级ORM，关联预加载逻辑会自动组装嵌套的模型实例，不需要手动处理联表返回的重复数据。
  • Tortoise-ORM：异步场景常用ORM，内置关联预加载和嵌套序列化能力，适配异步数据库驱动。
  • Django ORM：配合select_related、prefetch_related预加载关联数据，结合DRF的嵌套序列化器即可直接输出嵌套结构。

• 轻量转换工具（原生SQL场景适用）
  如果直接执行原生SQL拿到元组/字典列表，不想引入重量级ORM，可以选择以下轻量方案：

  • marshmallow系列工具：通过marshmallow定义各层级的字段和关联Schema，传入扁平化的查询结果后，工具会自动按主键去重、组装嵌套结构，支持字段过滤、自定义格式化，输出结果可直接序列化为JSON，marshmallow-sqlalchemy、jsonapi-marshmallow等扩展库还能直接对接SQLAlchemy模型减少重复定义。
  • Pydantic + 手动组装：针对固定结构的场景，手动写组装逻辑成本极低，核心思路是用字典按各层级主键做缓存，遍历行记录时逐层判断节点是否存在，不存在则初始化，存在则直接追加子项，最后用Pydantic做类型校验即可，没有额外依赖，性能远高于通用工具。
  • dataset：超轻量数据库工具库，内置nest方法，只需要指定各层级的关联键，就能快速把扁平表记录转成嵌套结构，适合3层以内的简单关联场景。

无依赖手动实现示例 #

针对最开始的学生-书籍两层关联，仅用标准库就能实现转换，代码如下：

def flat_to_nested(flat_records):
    student_cache = {}
    for row in flat_records:
        s_id, s_name, b_id, b_name = row
        # 初始化不存在的学生节点
        if s_id not in student_cache:
            student_cache[s_id] = {
                "id": s_id,
                "name": s_name,
                "books": {}  # 用字典存书籍，自动按book_id去重
            }
        # 追加书籍，同id书籍不会重复添加
        student_cache[s_id]["books"][b_id] = {"id": b_id, "name": b_name}
    
    # 把书籍字典转为列表，匹配目标输出格式
    result = []
    for student in student_cache.values():
        student["books"] = list(student["books"].values())
        result.append(student)
    return result

多层级关联只需要按照相同逻辑，逐层增加缓存判断即可，逻辑非常直观。

• 快速开发框架内置能力
  如果是做管理后台、低代码类项目，很多全栈框架已经内置了该能力：
  • FastAPI生态下的FastAPI-Amis-Admin，配置好模型关联关系后，接口会自动返回嵌套的关联数据，不需要手动写转换逻辑。
  • Flask生态下的Flask-Admin、Flask-RESTful扩展，配合模型关联定义也能自动完成嵌套结果组装。

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