如何将指定CSV转换为目标嵌套字典？已有Python代码未达预期

原始CSV数据 #

name,columns,tests
ABC_ESTIMATE_REFINED,cntquota,dbt_expectations.expect_column_to_exist
ABC_ESTIMATE_REFINED,cntquota,not_null
ABC_ESTIMATE_REFINED,is_purged,dbt_expectations.expect_column_to_exist
ABC_ESTIMATE_REFINED,is_purged,not_null

预期输出结构 #

{
    "name": "ABC_ESTIMATE_REFINED",
    "columns": [
        {
            "name": "cntquota",
            "tests": [
                "dbt_expectations.expect_column_to_exist",
                "not_null"
            ]
        },
        {
            "name": "is_purged",
            "tests": [
                "dbt_expectations.expect_column_to_exist",
                "not_null"
            ]
        }
    ]
}

原代码问题 #

用户原代码通过groupby结合apply生成的字典结构与预期不符，输出是多层嵌套的键值对，而非要求的包含name和columns的层级结构。

正确实现代码 #

import pandas as pd
import json

# 读取CSV数据
df = pd.read_csv('data.csv')

# 初始化结果字典
result = {}
# 获取唯一表名（假设所有行name字段值一致）
result['name'] = df['name'].iloc[0]

# 处理列与对应测试用例
columns_list = []
# 按columns字段分组遍历
for col_name, group in df.groupby('columns'):
    # 提取当前列的所有测试用例转为列表
    tests = group['tests'].tolist()
    columns_list.append({
        'name': col_name,
        'tests': tests
    })
result['columns'] = columns_list

# 格式化打印结果
print(json.dumps(result, indent=4))

代码说明 #

1. 读取CSV数据到DataFrame；
2. 初始化结果字典，直接取第一行的name值作为顶级键name的内容（假设所有行表名一致）；
3. 按columns字段分组，遍历每个分组构造列对应的测试用例字典，存入列表；
4. 将列列表赋值给结果字典的columns键；
5. 用json.dumps格式化输出，方便查看结构。

简洁实现版本 #

利用groupby和agg方法简化代码：

import pandas as pd
import json

df = pd.read_csv('data.csv')

# 按name分组，再聚合列对应的测试用例列表
grouped = df.groupby('name').apply(
    lambda x: x.groupby('columns')['tests'].agg(list).reset_index().rename(columns={'columns': 'name'})
).reset_index()

# 构造目标字典
result = {
    'name': grouped['name'].iloc[0],
    'columns': grouped[0].iloc[0].to_dict('records')
}

print(json.dumps(result, indent=4))

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