Python调用USDA API获取营养数据：分页机制详解

本文详细阐述了如何使用Python调用USDA食品数据中心（FDC）API，突破其默认50条数据的分页限制，完整获取营养数据。文章首先解析了API的分页机制，强调pageSize和pageNumber参数的重要性。随后，提供了一套实用的Python代码示例，展示了如何通过设置pageSize最大值、迭代请求不同页码、累积数据、以及运用requests.Session和错误处理机制，高效稳定地抓取完整数据集。此外，还分享了API Key安全、数据结构变化应对、API限流等注意事项与最佳实践。掌握这些技巧，开发者可以轻松应对分页API的数据获取挑战，为后续的营养数据分析与应用提供有力支持，并最终将数据导出至Excel，方便进一步处理。

本教程旨在解决使用USDA食品数据中心（FDC）API时遇到的数据分页限制问题，特别是默认仅返回50条结果的情况。文章将详细解释API分页机制，并通过Python代码示例演示如何有效地利用pageSize和pageNumber参数，结合请求会话和错误处理，实现完整数据集的迭代获取与处理，最终导出至Excel。

1. 理解USDA FDC API的分页机制

在使用任何外部API时，查阅其官方文档是至关重要的一步。对于USDA食品数据中心（FDC）API，当您通过foods/search端点查询数据时，可能会发现每次请求默认只返回50条结果，即使符合条件的数据远超此数量。这是因为API采用了分页（Pagination）机制来管理大量数据的传输。

API响应中通常会包含以下关键信息，指示分页状态：

• currentPage: 当前页码。
• totalPages: 总页数。
• pageSize: 每页返回的结果数量（默认为50）。

API文档明确指出，pageSize参数的默认值为50，但最大可以设置为200。同时，可以通过pageNumber参数指定要获取的页码。这意味着要获取所有符合条件的数据，我们需要执行一系列请求，每次请求不同的页码，直到所有页面都被遍历。

2. 实现完整数据获取的策略

为了克服50条结果的限制并获取完整数据集，我们需要采取以下策略：

1. 设置pageSize参数： 在初始请求中，将pageSize设置为API允许的最大值（例如200），以减少总的请求次数。
2. 获取总页数： 发送第一个请求后，从API响应中提取totalPages字段的值。
3. 迭代请求： 从第二页开始，循环遍历到totalPages，每次请求时更新pageNumber参数。
4. 累积数据： 将每次请求返回的数据累加到一个列表中。
5. 错误处理： 在请求过程中加入适当的错误处理机制，以应对网络问题或API响应异常。
6. 使用requests.Session： 对于多次向同一主机发出的请求，使用requests.Session可以提高效率，因为它会重用底层的TCP连接。

3. Python代码实现：迭代获取USDA营养数据

以下是一个优化后的Python函数，它封装了上述策略，能够从USDA FDC API完整地获取指定食物的营养数据。

import requests
import json
import pandas as pd
import time # 引入time模块用于模拟请求间隔

def get_all_food_data(query_name: str, api_key: str, max_page_size: int = 200) -> list:
    """
    通过USDA FDC API获取指定查询名称的所有食品数据，处理分页。

    Args:
        query_name (str): 查询的食物名称，例如 'raw'。
        api_key (str): 您的USDA API密钥。
        max_page_size (int): 每页最大结果数，默认为200。

    Returns:
        list: 包含所有食品数据的列表，每个元素是一个字典。
              如果获取失败，返回None。
    """
    all_foods = []
    base_url = "https://api.nal.usda.gov/fdc/v1/foods/search"

    # 使用requests.Session来重用TCP连接，提高效率
    with requests.Session() as session:
        # 构建初始请求URL，设置最大pageSize
        initial_url = f"{base_url}?api_key={api_key}&query={query_name}&pageSize={max_page_size}"

        try:
            print(f"开始获取 '{query_name}' 的数据...")
            # 发送第一个请求以获取总页数和第一页数据
            response = session.get(initial_url, timeout=15)
            response.raise_for_status()  # 检查HTTP响应状态码，如果不是200，则抛出异常

            api_response_json = response.json()

            # 检查API响应中是否存在'foods'键
            if 'foods' not in api_response_json or not api_response_json['foods']:
                print(f"未找到 '{query_name}' 的食品数据或API响应格式异常。")
                return []

            all_foods.extend(api_response_json["foods"])

            total_pages = api_response_json.get("totalPages", 1)
            print(f"总共找到 {api_response_json.get('totalHits', 0)} 条结果，分布在 {total_pages} 页。")

            # 从第二页开始循环获取所有剩余页面
            for page_num in range(2, total_pages + 1):
                page_url = f"{initial_url}&pageNumber={page_num}"
                print(f"正在获取第 {page_num}/{total_pages} 页数据...")

                # 建议在每次请求之间添加短暂的延迟，以避免API限流
                time.sleep(0.1) 

                page_response = session.get(page_url, timeout=15)
                page_response.raise_for_status()

                page_api_response_json = page_response.json()
                if 'foods' in page_api_response_json:
                    all_foods.extend(page_api_response_json["foods"])
                else:
                    print(f"警告：第 {page_num} 页API响应中未包含 'foods' 键。")

        except requests.exceptions.HTTPError as http_err:
            print(f"HTTP错误发生: {http_err} - URL: {response.url}")
            return None
        except requests.exceptions.ConnectionError as conn_err:
            print(f"连接错误发生: {conn_err}")
            return None
        except requests.exceptions.Timeout as timeout_err:
            print(f"请求超时: {timeout_err}")
            return None
        except requests.exceptions.RequestException as req_err:
            print(f"请求发生未知错误: {req_err}")
            return None
        except json.JSONDecodeError as json_err:
            print(f"JSON解析错误: {json_err} - 响应内容: {response.text[:200]}...")
            return None
        except Exception as e:
            print(f"发生意外错误: {e}")
            return None

    print(f"成功获取所有 {len(all_foods)} 条食品数据。")
    return all_foods

# --- 使用示例 ---
if __name__ == "__main__":
    YOUR_API_KEY = "YOUR_USDA_API_KEY"  # 替换为您的实际API密钥
    SEARCH_QUERY = "apple"  # 替换为您想查询的食物

    # 1. 调用函数获取所有食品数据
    food_items = get_all_food_data(SEARCH_QUERY, YOUR_API_KEY)

    if food_items:
        # 2. 将获取到的数据转换为Pandas DataFrame
        table_data = []
        for food_item in food_items:
            row = {
                "Description": food_item.get("description", "N/A"),
                "FDC_ID": food_item.get("fdcId", "N/A")
            }
            # 提取营养成分，注意有些食品可能没有所有营养成分
            for nutrient in food_item.get("foodNutrients", []):
                nutrient_name = nutrient.get("nutrientName")
                value = nutrient.get("value")
                if nutrient_name and value is not None:
                    row[nutrient_name] = value
            table_data.append(row)

        df = pd.DataFrame(table_data)

        # 3. 导出到Excel
        output_filename = f'{SEARCH_QUERY}_nutritional_facts.xlsx'
        try:
            df.to_excel(output_filename, index=False)
            print(f"数据已成功导出到 '{output_filename}'")
            print(f"DataFrame的最后一行数据:\n{df.iloc[-1]}")
        except Exception as e:
            print(f"导出Excel失败: {e}")
    else:
        print("未能获取食品数据，无法生成Excel文件。")

代码说明：

• get_all_food_data 函数：
  • 接收 query_name（查询关键词）和 api_key 作为参数。
  • 使用 requests.Session() 创建一个会话对象，以优化连续请求。
  • 构建基础URL，并首次请求时将 pageSize 设置为 max_page_size（默认为200）。
  • 通过 response.raise_for_status() 检查HTTP响应状态，如果不是2xx，则抛出 HTTPError。
  • 从第一个响应中提取 totalPages。
  • 使用 for 循环从第2页迭代到 totalPages，每次请求时通过 pageNumber 参数指定页码。
  • all_foods.extend() 用于将每一页获取到的食品数据追加到 all_foods 列表中。
  • 错误处理： 使用 try-except 块捕获 requests 库可能抛出的各种异常（如 HTTPError, ConnectionError, Timeout 等），增强代码的健壮性。
  • API限流考虑： 尽管USDA API文档未明确提及严格的速率限制，但在循环请求中添加 time.sleep(0.1) 是一种良好的实践，可以避免因请求过快而被服务器暂时阻止。
• 数据处理与导出：
  • if __name__ == "__main__": 块展示了如何调用 get_all_food_data 函数。
  • 获取到 food_items 列表后，通过遍历列表，将每个食品的描述和营养成分组织成字典，然后添加到 table_data 列表中。
  • 使用 pandas.DataFrame(table_data) 将列表转换为DataFrame。
  • 最后，df.to_excel(output_filename, index=False) 将DataFrame导出为Excel文件，index=False 避免将DataFrame的索引写入Excel。

4. 注意事项与最佳实践

• API Key安全： 永远不要在公开代码库中直接硬编码您的API密钥。在实际应用中，应将其存储在环境变量、配置文件或秘密管理服务中。
• 错误处理： 示例代码中包含了基本的错误处理，但在生产环境中，您可能需要更详细的日志记录、重试机制或用户友好的错误提示。
• 数据结构变化： API响应的数据结构可能会随着版本更新而变化。在处理 foodNutrients 时，建议使用 get() 方法安全地访问字典键，以防止因键不存在而引发错误。
• API限流： 尽管在示例中加入了 time.sleep()，但如果您的应用程序需要处理大量请求，应仔细查阅API文档关于速率限制（Rate Limiting）的规定，并相应地调整请求间隔或采用令牌桶等限流策略。
• 数据清洗： 从API获取的原始数据可能需要进一步的清洗和标准化，例如处理缺失值、统一单位等，这通常在Pandas DataFrame创建之后进行。

5. 总结

通过本教程，我们深入理解了USDA FDC API的分页机制，并学习了如何通过Python编写健壮的代码来完整地获取所有符合条件的营养数据。核心在于利用API提供的 pageSize 和 pageNumber 参数进行迭代请求，同时结合 requests.Session 提高效率，并加入全面的错误处理。掌握这一技巧，您将能够更有效地从分页API中提取所需的所有信息，为后续的数据分析和应用打下坚实基础。

以上就是《Python调用USDA API获取营养数据：分页机制详解》的详细内容，更多关于的资料请关注golang学习网公众号！