Python对接Asana API提取数据求助：数据工程前置环节遇困

一、优先使用官方 Asana Python SDK #

直接用 requests 调用 API 容易遇到认证、分页、字段缺失等问题，官方 SDK 已封装这些逻辑，能大幅降低开发成本：

• 安装 SDK：pip install asana
• 准备认证：在 Asana 账户设置中生成个人访问令牌 (PAT)
• 基础示例（获取项目任务）：

import asana

# 初始化客户端
client = asana.Client.access_token("你的Asana_PAT")

# 获取指定项目的任务数据（指定需要的字段）
project_id = "你的项目ID"
tasks = client.tasks.find_by_project(
    project_id, 
    opt_fields=["name", "assignee", "due_on", "completed"]
)

# 转换为列表方便后续处理
task_list = list(tasks)

二、Excel 数据提取与 Asana 同步 #

用 pandas 处理 Excel 数据，再通过 SDK 同步到 Asana：

• 安装依赖：pip install pandas openpyxl
• 示例代码：

import pandas as pd

# 读取 Excel 任务数据
df = pd.read_excel("待同步任务.xlsx")

# 遍历数据创建 Asana 任务
workspace_id = "你的工作区ID"
for _, row in df.iterrows():
    client.tasks.create_in_workspace(
        workspace_id,
        {
            "name": row["任务名称"],
            "assignee": row["负责人邮箱"],
            "due_on": row["截止日期"].strftime("%Y-%m-%d") if pd.notna(row["截止日期"]) else None,
            "projects": [project_id]
        }
    )

三、解决 requests 调用的常见问题 #

如果必须用 requests，针对常见问题给出修复方案：

• 认证失败：请求头必须包含 Authorization: Bearer 你的PAT
• 分页遗漏数据：Asana API 采用分页返回，需循环获取 next_page 链接的内容
• 字段不全：通过 opt_fields 参数指定需要返回的字段

示例修复后的 requests 代码：

import requests

headers = {
    "Authorization": "Bearer 你的Asana_PAT",
    "Content-Type": "application/json"
}

# 初始请求
url = f"https://app.asana.com/api/1.0/projects/{project_id}/tasks"
params = {"opt_fields": "name,assignee,due_on,completed"}
response = requests.get(url, headers=headers, params=params)

# 处理分页，获取全部数据
all_tasks = []
while True:
    data = response.json()
    all_tasks.extend(data["data"])
    next_page = data.get("next_page")
    if not next_page:
        break
    response = requests.get(next_page["uri"], headers=headers)

四、后续数据工程处理 #

将获取的 Asana 数据转换为 DataFrame 后，可进行清洗、分析：

import pandas as pd

# 转换为 DataFrame
df_asana = pd.DataFrame(task_list)

# 数据清洗：处理缺失值、提取嵌套字段
df_asana["due_on"] = pd.to_datetime(df_asana["due_on"])
df_asana["负责人姓名"] = df_asana["assignee"].apply(lambda x: x["name"] if x else None)

# 保存清洗后的数据
df_asana.to_csv("Asana任务数据_清洗后.csv", index=False)

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