如何用Python/Pandas将SQL联表结果转为三层嵌套字典？

我从三表关联的SQL存储过程中得到了如下扁平数据：

data = [
    {"so_number": "ABC", "po_status": "OPEN", "item_id": 0, "part_number": "XTZ", "ticket_id": 10, "ticket_month": "JUNE"},
    {"so_number": "ABC", "po_status": "OPEN", "item_id": 0, "part_number": "XTZ", "ticket_id": 11, "ticket_month": "JUNE"},
    {"so_number": "ABC", "po_status": "OPEN", "item_id": 1, "part_number": "XTY", "ticket_id": 12, "ticket_month": "JUNE"},
    {"so_number": "DEF", "po_status": "OPEN", "item_id": 3, "part_number": "XTU", "ticket_id": 13, "ticket_month": "JUNE"},
    {"so_number": "DEF", "po_status": "OPEN", "item_id": 3, "part_number": "XTU", "ticket_id": 14, "ticket_month": "JUNE"},
    {"so_number": "DEF", "po_status": "OPEN", "item_id": 3, "part_number": "XTU", "ticket_id": 15, "ticket_month": "JUNE"}
]

需要按so_number和item_id分组，转换成如下嵌套结构的字典列表：

[
  {
    "so_number": "ABC",
    "po_status": "OPEN",
    "line_items": [
      {
        "item_id": 0,
        "part_number": "XTZ",
        "tickets": [
          {"ticket_id": 10, "ticket_month": "JUNE"},
          {"ticket_id": 11, "ticket_month": "JUNE"}
        ]
      },
      {
        "item_id": 1,
        "part_number": "XTY",
        "tickets": [{"ticket_id": 12, "ticket_month": "JUNE"}]
      }
    ]
  },
  {
    "so_number": "DEF",
    "po_status": "OPEN",
    "line_items": [
      {
        "item_id": 3,
        "part_number": "XTU",
        "tickets": [
          {"ticket_id": 13, "ticket_month": "JUNE"},
          {"ticket_id": 14, "ticket_month": "JUNE"},
          {"ticket_id": 15, "ticket_month": "JUNE"}
        ]
      }
    ]
  }
]

不想通过循环访问三张SQL表来生成（效率低且非最佳实践），求高效实现方式（可使用Pandas）。

一、原生Python实现（无需第三方库） #

核心思路是用字典做分组容器，一次遍历完成多层分组，时间复杂度O(n)，效率很高。

def transform_data(data):
    so_map = {}
    for row in data:
        so_num = row["so_number"]
        # 处理SO层级
        if so_num not in so_map:
            so_map[so_num] = {
                "so_number": so_num,
                "po_status": row["po_status"],
                "line_items": {}
            }
        so_entry = so_map[so_num]
        
        item_id = row["item_id"]
        # 处理item层级
        if item_id not in so_entry["line_items"]:
            so_entry["line_items"][item_id] = {
                "item_id": item_id,
                "part_number": row["part_number"],
                "tickets": []
            }
        item_entry = so_entry["line_items"][item_id]
        
        # 添加ticket
        item_entry["tickets"].append({
            "ticket_id": row["ticket_id"],
            "ticket_month": row["ticket_month"]
        })
    
    # 把line_items的字典转成列表
    result = []
    for so_entry in so_map.values():
        so_entry["line_items"] = list(so_entry["line_items"].values())
        result.append(so_entry)
    return result

# 调用示例
output = transform_data(data)
print(output)

二、Pandas实现（适合大数据量场景） #

利用Pandas的groupby和聚合函数，结合to_dict完成转换，代码更简洁，处理大规模数据时性能更优。

import pandas as pd

df = pd.DataFrame(data)

# 先按so_number和item_id分组，聚合ticket数据
agg_result = df.groupby(["so_number", "po_status", "item_id", "part_number"]).apply(
    lambda x: x[["ticket_id", "ticket_month"]].to_dict("records")
).reset_index(name="tickets")

# 再按so_number和po_status分组，聚合line_items数据
final_result = agg_result.groupby(["so_number", "po_status"]).apply(
    lambda x: x[["item_id", "part_number", "tickets"]].to_dict("records")
).reset_index(name="line_items")

# 转换为目标字典列表
output = final_result.to_dict("records")
print(output)

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