Python调用OpenAlex API提取文献数据时处理空值报错优化

问题背景 #

通过OpenAlex API提取学术出版物数据时，部分文献的institutions字段为空列表[]，而非包含机构对象的列表，导致代码触发IndexError: list index out of range。需要在缺失信息的位置填入None，同时保证代码高效处理数十万级请求。现有代码无法处理如magid为2301544176的不完整条目，相关代码及报错栈如下：

原代码 #

import requests
import json

baseurl = 'https://api.openalex.org/works?filter=ids.mag:'

#**upper magid_listworks without problems**
#magid_list = [2301543590, 2301543835]

#**error occur**
#**see page "https://api.openalex.org/works?filter=ids.mag:2301544176" no information for institution given**
magid_list = [2301543590, 2301543835, 2301544176]

def main_request(baseurl, endpoint):
    r = requests.get(baseurl + endpoint)
    return r.json()

def parse_json(response):
    charlist = []
    pupdate = data['results'][0]['publication_date']
    display_name = data['results'][0]['display_name']
    for item in response['results'][0]['authorships']:
        char = {
        'magid': str(x),
        'display_name': display_name,
        'pupdate': pupdate,
        'author': item['author']['display_name'],
        'institution_id': item['institutions'][0]['id']
        }
        
        charlist.append(char)
    return charlist

finallist = []

for x in magid_list:
    print(x)
    data = main_request(baseurl, str(x))
    finallist.extend(parse_json(main_request(baseurl, str(x))))

df = pd.DataFrame(finallist)

print(df.head(), df.tail())

报错栈 #

---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
f:\AlexPE\__programming\Masterarbeit.ipynb Cell 153 in <cell line: 37>()
     37 for x in list:
     38     print(x)
---&gt; 39     finallist.extend(parse_json(main_request(baseurl, str(x))))
     41 df = pd.DataFrame(finallist)
     43 #data = main_request(baseurl, endpoint)
     44 #print(get_pages(data))
     45 #print(parse_json(data))

f:\AlexPE\__programming\Masterarbeit.ipynb Cell 153 in parse_json(response)
     20 display_name = data['results'][0]['display_name']
     23 for item in response['results'][0]['authorships']:
     24     char = {
     25     'magid': str(x),
     26     'display_name': display_name,
     27     'pupdate': pupdate,
     28     'author': item['author']['display_name'],
---&gt; 29     'institution_id': item['institutions'][0]['id']
     30     }
     32     charlist.append(char)
     33 return charlist

IndexError: list index out of range 

错误原因 #

1. 直接索引空列表：代码中item['institutions'][0]['id']直接访问空列表的第0位，当institutions为空时触发IndexError。
2. 变量引用错误：parse_json函数中使用全局变量data和x，而非传入参数，逻辑混乱。
3. 重复请求API：循环中对同一个magid调用两次main_request，浪费资源且降低效率。
4. 无异常处理：未处理请求失败、字段缺失等情况，无法应对大规模请求的稳定性问题。

修复后代码 #

import requests
import pandas as pd
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 配置请求会话与重试机制，提升大规模请求稳定性
session = requests.Session()
retry_strategy = Retry(
    total=5,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)

baseurl = 'https://api.openalex.org/works?filter=ids.mag:'

# 测试用magid列表，包含问题条目
magid_list = [2301543590, 2301543835, 2301544176]

def main_request(magid):
    """发送API请求，返回解析后的JSON数据，处理请求异常"""
    url = f"{baseurl}{magid}"
    try:
        response = session.get(url, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"magid {magid} 请求失败: {str(e)}")
        return None

def parse_result(magid, result_data):
    """解析单条结果数据，安全处理缺失字段"""
    records = []
    pub_date = result_data.get('publication_date')
    title = result_data.get('display_name')
    
    for authorship in result_data.get('authorships', []):
        author_name = authorship.get('author', {}).get('display_name')
        # 处理空institutions列表，缺失时填充None
        institutions = authorship.get('institutions', [])
        inst_id = institutions[0].get('id') if institutions else None
        
        record = {
            'magid': str(magid),
            'title': title,
            'publication_date': pub_date,
            'author': author_name,
            'institution_id': inst_id
        }
        records.append(record)
    return records

final_records = []

for magid in magid_list:
    print(f"处理magid: {magid}")
    api_data = main_request(magid)
    if api_data and api_data.get('results'):
        # 按magid过滤仅返回一条结果
        single_result = api_data['results'][0]
        final_records.extend(parse_result(magid, single_result))

# 转换为DataFrame
df = pd.DataFrame(final_records)
print("前5条数据:")
print(df.head())
print("\n后5条数据:")
print(df.tail())

关键优化点 #

• 安全访问字段：使用get()方法访问嵌套字典/列表，避免KeyError；判断institutions是否为空，为空则设institution_id为None。
• 减少重复请求：每个magid仅请求一次API，复用返回数据。
• 请求稳定性：使用requests.Session和重试机制，自动处理限流、网络波动等问题，适合数十万级请求。
• 异常捕获：处理请求失败情况，避免程序崩溃，同时输出错误日志便于排查。

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