Python调用BetterImpact API导出CSV时出现KeyError问题排查求助

问题描述 #

我正在运行一段Python代码，调用BetterImpact的API提取数据并导出至CSV文件。目前CSV文件可以生成，但运行时抛出KeyError错误。麻烦帮我排查一下可能的问题所在？

错误类型 #

KeyError

代码 #

import http.client
import json
import csv
import os

conn = http.client.HTTPSConnection("api.betterimpact.com")
conn1 = http.client.HTTPSConnection("api.betterimpact.com")

if os.path.exists("CSVVolunteerOutput.csv"):
    os.remove("CSVVolunteerOutput.csv")

headers = {
    'Authorization': 'Basic XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX',
    'Cookie': '; TrackingClientId=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
}

conn.request("GET", "/v1/enterprise/users/?volunteer_status=accepted&include_custom_fields=false&include_qualifications=false&page_Size=250&include_verified_volunteers_background_check_results=false", headers=headers)
res = conn.getresponse()
data = json.load(res)

conn1.request("GET", "/v1/enterprise/users/?volunteer_status=accepted&include_custom_fields=false&include_qualifications=false&page_Size=250&include_verified_volunteers_background_check_results=false&page_number=0", headers = headers)
res1 = conn1.getresponse()
data1 = json.load(res1)

if data == None or data == "" or len(data) == 0:
    print("Check API Credentials..")
    exit()

volunteer_status = "Accepted"
pageNum = 0
_page_count = data1['header']['page_count']

while True:
    pageNum+=1
    with open('CSVVolunteerOutput.csv', 'a', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(["SyncID", "FirstName", "LastName", "Jobtitle", "Division", "BusinessUnit", "RegionArea", "WorkLocation", "Department", "WorkEmailAddressPrimary", "PersonalEmailAddress", "PersonalMobilePhonePrimary", "WorkCountry"])
        for user in data['users']:
            _id = user['user_id']
            _firstName = user['first_name']
            _surName = user['last_name']
            _emailAddress = user['email_address']
            _emailAddressSec = user['secondary_email_address']
            _cellPhone = user['cell_phone']
            _country = user['country']
            for details in user['memberships']:
                _orgName = details['organization_name']
                _volunteerStatus = details['volunteer_status']
                if volunteer_status == _volunteerStatus:
                    writer.writerow([_id, _firstName, _surName, "Volunteer", "", "", "", _orgName, "", _emailAddress, _emailAddressSec, _cellPhone, _country])
    if pageNum > int(_page_count):
        break
    else:
        conn.request("GET", "/v1/enterprise/users/?volunteer_status=accepted&include_custom_fields=false&include_qualifications=false&page_Size=250&include_verified_volunteers_background_check_results=false&page_number="+str(pageNum), headers=headers)
        res = conn.getresponse()
        data = json.load(res)

print("CSV file created successfully")

可能的问题与修复方案 #

KeyError本质是你在尝试访问字典中不存在的键，结合你的代码逻辑，大概率是API返回的JSON结构和你预期的不一致，以下是几个高频出错点的排查和解决办法：

1. 总页数获取时的键缺失 #

你通过data1['header']['page_count']获取总页数，但如果API响应里没有header键，或者header下没有page_count，就会触发错误。

• 排查：在赋值前打印data1的完整结构，确认键是否存在：

  print(json.dumps(data1, indent=2))
  

• 修复：用字典的get()方法安全取值，给缺失的键设置默认值：

  _page_count = data1.get('header', {}).get('page_count', 0)
  

2. 用户列表的键缺失 #

如果API返回的响应中没有users键（比如请求失败返回错误信息），data['users']会直接报错。

• 排查：进入循环前打印data内容，确认是否包含users数组：

  print(json.dumps(data, indent=2))
  

• 修复：安全获取用户列表，无数据时跳过当前页面：

  users = data.get('users', [])
  if not users:
      print(f"Page {pageNum} has no volunteer data")
      continue
  for user in users:
      # 后续处理逻辑
  

3. 用户属性或会员信息的键缺失 #

部分用户数据可能缺失secondary_email_address、cell_phone或memberships这类字段，直接通过键访问会触发错误。

• 排查：捕获KeyError并打印出错的用户数据：

  try:
      _id = user['user_id']
      # 其他属性赋值
  except KeyError as e:
      print(f"Missing key {e} in user data: {json.dumps(user, indent=2)}")
      continue
  

• 修复：统一用get()方法获取属性，给缺失字段设置空字符串默认值：

  _id = user.get('user_id', '')
  _firstName = user.get('first_name', '')
  _surName = user.get('last_name', '')
  _emailAddress = user.get('email_address', '')
  _emailAddressSec = user.get('secondary_email_address', '')
  _cellPhone = user.get('cell_phone', '')
  _country = user.get('country', '')
  # 处理会员信息，无数据时返回空列表
  memberships = user.get('memberships', [])
  for details in memberships:
      _orgName = details.get('organization_name', '')
      _volunteerStatus = details.get('volunteer_status', '')
      if volunteer_status == _volunteerStatus:
          writer.writerow([_id, _firstName, _surName, "Volunteer", "", "", "", _orgName, "", _emailAddress, _emailAddressSec, _cellPhone, _country])
  

4. 分页请求的响应异常 #

循环请求后续页面时，可能遇到API返回错误状态码（比如401权限问题、404页面不存在），此时解析的JSON结构会不符合预期。

• 排查：每次获取响应后先检查状态码：

  res = conn.getresponse()
  print(f"Page {pageNum} response status: {res.status}")
  if res.status != 200:
      print(f"Error content: {res.read().decode()}")
      continue
  data = json.load(res)
  

• 修复：仅在响应状态正常时处理数据，避免解析错误响应导致的KeyError。

额外优化建议 #

• 你创建了两个重复的HTTPS连接conn和conn1，可以复用一个连接减少资源消耗；
• 表头writer.writerow(...)放在循环里会导致每个页面都写入一次表头，建议把表头写入逻辑放在循环外，仅执行一次：

  # 先写入表头
  with open('CSVVolunteerOutput.csv', 'w', newline='') as file:
      writer = csv.writer(file)
      writer.writerow(["SyncID", "FirstName", "LastName", "Jobtitle", "Division", "BusinessUnit", "RegionArea", "WorkLocation", "Department", "WorkEmailAddressPrimary", "PersonalEmailAddress", "PersonalMobilePhonePrimary", "WorkCountry"])
  
  # 再循环处理分页数据并追加写入
  while True:
      pageNum+=1
      with open('CSVVolunteerOutput.csv', 'a', newline='') as file:
          writer = csv.writer(file)
          # 处理用户数据写入
  

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