将Asana任务数据转换为Pandas DataFrame以导入Power BI

我是Python新手，这是我的第一个项目，目的是将Asana的任务数据提取后导入Power BI。由于没有Asana企业版许可，无法使用官方连接器，因此选择用Python实现。目前已成功复制调整Asana提供的代码获取到任务数据，但不知道如何将这些嵌套结构的数据转换为动态Pandas DataFrame，以便导入Power BI。已导入pandas库。

现有Asana数据提取代码 #

import asana
from asana.rest import ApiException
from pprint import pprint

configuration = asana.Configuration()
configuration.access_token = 'Abcd123'
api_client = asana.ApiClient(configuration)

tasks_api_instance = asana.TasksApi(api_client)
opts = {
    'project': "123456789",
    'opt_fields': "name,actual_time_minutes,assignee,assignee.name,"
                  "custom_fields,custom_fields.name,custom_fields.display_value,"
                  "due_on,start_on,"
                  "permalink_url"
}

try:
    tasks = tasks_api_instance.get_tasks(opts)
    for task in tasks:
        pprint(task)
except ApiException as e:
    print("Exception when calling TasksApi->get_tasks: %s\n" % e)

返回的数据示例 #

{'actual_time_minutes': None,
 'assignee': {'gid': '1104899036650576', 'name': 'Jane Doe'},
 'custom_fields': [{'display_value': None,
                    'gid': '1207268873906857',
                    'name': 'Estimated time'},
                   {'display_value': None,
                    'gid': '1207268873683694',
                    'name': 'Priority'},
                   {'display_value': 'Closed',
                    'gid': '1207268873683699',
                    'name': 'Status'},
                   {'display_value': 'https://dev.azure.com/clientsite',
                    'gid': '1207269260215808',
                    'name': 'Link to DevOps'},
                   {'display_value': None,
                    'gid': '1207277291413677',
                    'name': 'Sprint'},
                   {'display_value': '4265',
                    'gid': '1207269541980519',
                    'name': 'DevOps ID'}],
 'due_on': '2024-04-05',
 'gid': '1207277618607073',
 'name': 'Update Javascript on this Thing',
 'start_on': None}
{'actual_time_minutes': None,
 'assignee': None,
 'custom_fields': [{'display_value': None,
                    'gid': '1207268873906857',
                    'name': 'Estimated time'},
                   {'display_value': None,
                    'gid': '1207268873683694',
                    'name': 'Priority'},
                   {'display_value': 'On Hold',
                    'gid': '1207268873683699',
                    'name': 'Status'},
                   {'display_value': 'https://dev.azure.com/clientsite',
                    'gid': '1207269260215808',
                    'name': 'Link to DevOps'},
                   {'display_value': None,
                    'gid': '1207277291413677',
                    'name': 'Sprint'},
                   {'display_value': '3867',
                    'gid': '1207269541980519',
                    'name': 'DevOps ID'}],
 'due_on': '2023-12-12',
 'gid': '1207277735170286',
 'name': 'Create Workflow to Do This',
 'start_on': None}

需求 #

需要提取的字段包括：name、assignee.name、Status、Estimated time、actual_time_minutes，将每条记录转换为DataFrame的一行，且需适配动态变化的返回数据，不能硬编码固定值。已能创建测试DataFrame，但不知如何适配Asana的数据。

核心思路 #

遍历每条任务数据，将嵌套结构（如assignee子字段、custom_fields数组）转换为扁平的键值对，统一收集后生成DataFrame，确保代码能适配字段顺序或数量变化的情况。

完整实现代码 #

import asana
from asana.rest import ApiException
import pandas as pd

configuration = asana.Configuration()
configuration.access_token = 'Abcd123'
api_client = asana.ApiClient(configuration)

tasks_api_instance = asana.TasksApi(api_client)
opts = {
    'project': "123456789",
    'opt_fields': "name,actual_time_minutes,assignee,assignee.name,"
                  "custom_fields,custom_fields.name,custom_fields.display_value,"
                  "due_on,start_on,"
                  "permalink_url"
}

# 存储所有任务的扁平数据
task_data_list = []

try:
    tasks = tasks_api_instance.get_tasks(opts)
    for task in tasks:
        # 初始化当前任务的扁平字典
        flat_task = {}
        
        # 提取直接字段：name、actual_time_minutes
        flat_task['name'] = task.get('name')
        flat_task['actual_time_minutes'] = task.get('actual_time_minutes')
        
        # 提取assignee.name，兼容assignee为None的情况
        flat_task['assignee.name'] = task.get('assignee', {}).get('name')
        
        # 动态提取custom_fields中的指定字段
        custom_fields = task.get('custom_fields', [])
        for field in custom_fields:
            field_name = field.get('name')
            if field_name in ['Status', 'Estimated time']:
                flat_task[field_name] = field.get('display_value')
        
        # 将处理后的任务数据加入列表
        task_data_list.append(flat_task)
        
    # 转换为DataFrame
    df = pd.DataFrame(task_data_list)
    print(df)
    
    # 可选：导出为CSV，直接导入Power BI
    df.to_csv('asana_tasks.csv', index=False)
    
except ApiException as e:
    print("Exception when calling TasksApi->get_tasks: %s\n" % e)

代码细节解释 #

1. 扁平数据收集：用task_data_list存储每条任务的扁平字典，最后一次性转换为DataFrame，效率更高。
2. 安全提取字段：使用dict.get()方法，即使字段不存在也不会报错，返回None，避免程序崩溃。
3. 嵌套字段处理：对assignee字段，用task.get('assignee', {})确保即使assignee为None，后续的.get('name')也能正常执行。
4. 动态处理custom_fields：遍历custom_fields数组，只提取指定名称的字段，无需硬编码数组索引，适配字段顺序或数量变化的场景。
5. Power BI适配：导出的CSV文件可以直接在Power BI中通过"获取数据"功能导入，无需额外处理。

输出示例 #

运行代码后会生成如下结构的DataFrame：

name  actual_time_minutes assignee.name  Status Estimated time
0  Update Javascript on this Thing                  None      Jane Doe  Closed           None
1    Create Workflow to Do This                  None          None  On Hold           None
2  Write the Script for this Thing                  None          None     New           None
3    Go Do This Important Thing                  None      John Doe  Closed           None

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