如何将PostgreSQL数据插入Python嵌套字典并按类型统计计数

问题背景 #

需要将PostgreSQL返回的四列数据（coe、coe_type、count、coe_status）转换为指定结构的嵌套字典，要求每个作者（coe值）下的Open、Closed、All三类，分别统计Published和Non-Published的计数。当前实现仅All节点有数据，Open、Closed节点为空，需修正逻辑。

示例数据 #

coe,  coe_type, count, coe_status
Author 1, Open,       10,    Published
Author 2, Closed,     20,    Not-Published
Author 1, Closed,     15,    Published
Author 1, Open,       5,     Not-Published

当前错误返回结构 #

{
  "data": {
    "Author 1": {
      "Open": {},
      "Closed": {},
      "All": {
        "Published": 1,
        "Non-Published": 1
      }
    }
  }
}

期望返回结构 #

{
  "data": {
    "Author 1": {
      "Open": {
        "Published": 10,
        "Non-Published": 5
      },
      "Closed": {
        "Published": 15,
        "Non-Published": 0
      },
      "All": {
        "Published": 25,
        "Non-Published": 5
      }
    },
    "Author 2": {
      "Open": {
        "Published": 0,
        "Non-Published": 0
      },
      "Closed": {
        "Published": 0,
        "Non-Published": 20
      },
      "All": {
        "Published": 0,
        "Non-Published": 20
      }
    }
  }
}

原代码问题分析 #

1. 变量命名混乱：category_headers包含无关值，book_category未定义，book_type_headers是列表却直接与字符串比较（book_type_headers == 'Published'）
2. 循环逻辑错误：每次处理行时都会重置Open/Closed/All为空字典，覆盖之前的统计值
3. 状态判断错误：未正确区分coe_status的取值，且未对All类型进行累加

修正后的代码 #

from collections import defaultdict

# 定义固定的类型和状态常量
BOOK_TYPES = ["Open", "Closed", "All"]
STATUS_TYPES = ["Published", "Non-Published"]

# 初始化嵌套结构：作者 -> 类型 -> 状态 -> 计数（默认值0）
response_result = defaultdict(lambda: {
    book_type: {status: 0 for status in STATUS_TYPES} 
    for book_type in BOOK_TYPES
})

# 执行SQL查询并处理结果
response = execute_query(sql_query, transaction_id, False)
for row in response:
    # 解析每行数据（若返回的是字符串格式，可先用ast.literal_eval转成列表）
    current_author, coe_type, count, coe_status = row
    count = int(count)
    
    # 统一状态名称（适配数据库返回的"Not-Published"）
    status = "Non-Published" if coe_status == "Not-Published" else coe_status
    
    # 更新对应类型的计数
    response_result[current_author][coe_type][status] += count
    # 同步更新All类型的计数
    response_result[current_author]["All"][status] += count

# 转换为最终的字典格式
final_data = {"data": dict(response_result)}

关键修正点 #

1. 使用defaultdict简化嵌套结构初始化，确保每个作者的Open/Closed/All都默认包含两个状态的初始值0
2. 统一状态名称处理，避免数据库返回值与期望格式不一致
3. 处理每行数据时，同时更新对应coe_type和All类型的状态计数
4. 移除原代码中无效的循环和错误比较逻辑，直接基于行数据精准更新

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