深入理解与实现USDA FDC API数据分页获取（分页.获取.理解.数据.API...）

本文旨在解决通过Python API获取USDA FDC营养数据时遇到的结果限制问题。我们将详细讲解API分页机制，介绍如何利用pageSize和pageNumber参数，并通过迭代请求实现完整数据集的检索，确保用户能够高效、准确地获取所有目标数据。1. 理解API分页机制

在使用第三方API时，特别是涉及大量数据检索的场景，API通常会采用分页（Pagination）机制来管理和限制单次请求返回的数据量。这种机制有几个主要目的：

• 减轻服务器压力： 避免一次性返回海量数据，消耗过多服务器资源。
• 优化网络传输： 减少单次请求的数据传输量，提高响应速度。
• 提升客户端性能： 客户端无需一次性处理所有数据，可以按需加载，提升用户体验。

对于USDA Food Data Central (FDC) API，默认情况下，其搜索接口（/foods/search）每次请求最多返回50条结果。如果未正确处理分页，用户将只能获取到数据集的冰山一角。

2. 识别API分页信息

要正确处理API分页，首先需要识别API响应中包含的分页信息。通常，API会在响应体中提供当前页码、总页数以及每页大小等关键信息。以USDA FDC API为例，当您发起一个搜索请求并获取到JSON响应时，会发现类似currentPage和totalPages这样的字段，它们明确指示了结果是分页的，并且提供了遍历所有页面所需的信息。

此外，查阅API官方文档是理解分页机制最直接有效的方式。USDA FDC API的文档会明确指出哪些参数用于控制分页，例如pageSize（每页返回结果数量）和pageNumber（请求的页码）。根据文档，pageSize的默认值是50，但最大可以设置为200。

3. 实现完整数据检索：分页请求策略

要获取所有符合条件的数据，我们需要实施一个迭代请求策略，即：

1. 发起第一次请求，获取第一页数据，并识别总页数。
2. 根据总页数，循环发起后续请求，每次请求不同的页码。
3. 将所有页面的数据合并起来，形成完整的查询结果。

以下是一个使用Python requests库实现USDA FDC API分页数据检索的示例：

import requests
import json
import pandas as pd

def get_all_food_data(query_term: str, api_key: str, max_page_size: int = 200) -> list:
    """
    通过迭代请求USDA FDC API，获取指定查询词的所有食物营养数据。

    Args:
        query_term (str): 搜索的食物名称或关键词。
        api_key (str): 您的USDA FDC API密钥。
        max_page_size (int): 每页请求的最大结果数量，USDA FDC API最大支持200。

    Returns:
        list: 包含所有食物营养数据的列表，每个元素是一个字典。
              如果请求失败或无数据，则返回空列表。
    """
    all_foods = []
    # 构建基础URL，设置API密钥和查询词，并指定最大页大小
    base_url = f"https://api.nal.usda.gov/fdc/v1/foods/search?api_key={api_key}&query={query_term}&pageSize={max_page_size}"

    with requests.Session() as session:  # 使用requests.Session保持连接，提高效率
        try:
            # 1. 获取第一页数据，并确定总页数
            print(f"Fetching page 1 for query: '{query_term}'...")
            response = session.get(base_url, timeout=15)
            response.raise_for_status()  # 检查HTTP请求是否成功
            api_response = response.json()

            if not api_response.get("foods"):
                print(f"No foods found for query: '{query_term}'.")
                return []

            all_foods.extend(api_response["foods"])
            total_pages = api_response.get("totalPages", 1)
            print(f"Total pages to retrieve: {total_pages}")

            # 2. 迭代获取后续页面数据
            for page_num in range(2, total_pages + 1):
                print(f"Fetching page {page_num}...")
                page_url = f"{base_url}&pageNumber={page_num}"
                page_response = session.get(page_url, timeout=15)
                page_response.raise_for_status()
                page_data = page_response.json()
                all_foods.extend(page_data["foods"])

        except requests.exceptions.HTTPError as http_err:
            print(f"HTTP error occurred: {http_err} - Status Code: {response.status_code}")
            return []
        except requests.exceptions.ConnectionError as conn_err:
            print(f"Connection error occurred: {conn_err}")
            return []
        except requests.exceptions.Timeout as timeout_err:
            print(f"Timeout error occurred: {timeout_err}")
            return []
        except requests.exceptions.RequestException as req_err:
            print(f"An unexpected error occurred: {req_err}")
            return []
        except json.JSONDecodeError:
            print(f"Failed to decode JSON from response: {response.text[:200]}...")
            return []

    return all_foods

# --- 示例使用 ---
if __name__ == "__main__":
    # 请替换为您的实际API密钥
    YOUR_API_KEY = "YOUR_USDA_API_KEY"
    search_query = "raw" # 例如，搜索所有“生”的食物

    # 1. 获取所有食物数据
    food_items = get_all_food_data(search_query, YOUR_API_KEY)

    if food_items:
        print(f"\nSuccessfully retrieved {len(food_items)} food items.")

        # 2. 将获取到的数据转换为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")
                if nutrient_name:
                    row[nutrient_name] = nutrient.get("value")
            table_data.append(row)

        df = pd.DataFrame(table_data)

        # 3. 数据清洗和处理（示例）
        # 移除完全为空的列（如果有很多稀疏营养素）
        df.dropna(axis=1, how='all', inplace=True)
        # 填充NaN值，例如用0或平均值
        df.fillna(0, inplace=True)

        print("\nDataFrame Head:")
        print(df.head())
        print(f"\nDataFrame Shape: {df.shape}")

        # 4. 导出到Excel
        try:
            output_filename = f'{search_query}_food_data.xlsx'
            df.to_excel(output_filename, index=False)
            print(f"\nData successfully exported to '{output_filename}'")
        except Exception as e:
            print(f"Error exporting to Excel: {e}")
    else:
        print("No data to process or export.")

4. 代码解析与注意事项

1. requests.Session()： 使用requests.Session()可以有效地复用TCP连接，对于需要进行多次HTTP请求的场景（如分页），这能显著提高性能。
2. pageSize参数： 在构建初始URL时，通过&pageSize=200将每页结果数量设置为最大值，以减少总请求次数。
3. totalPages识别： 第一次请求后，从API响应中提取totalPages字段，这是控制循环的关键。
4. 循环迭代： 使用range(2, total_pages + 1)从第二页开始遍历到最后一页，每次请求通过&pageNumber={page_num}指定页码。
5. 数据合并： 每次请求获取到的foods列表都通过all_foods.extend()方法添加到总列表中。
6. 错误处理：
   • response.raise_for_status()：这是一个非常重要的调用，它会在HTTP请求返回非200状态码时（如404 Not Found, 500 Internal Server Error等）抛出HTTPError异常。
   • try-except块：捕获不同类型的requests异常（如HTTPError, ConnectionError, Timeout, RequestException）以及json.JSONDecodeError，增强代码的健壮性。
   • 超时设置： 在session.get()中添加timeout参数，防止请求长时间无响应导致程序卡死。
7. 数据后处理：
   • 获取所有数据后，您可以像原问题中那样，将其转换为pandas.DataFrame进行进一步的清洗、分析和导出。
   • 注意处理food_item.get("description", "N/A")等，使用.get()方法可以避免键不存在时抛出KeyError。
   • 根据实际需求，可能需要对DataFrame进行填充缺失值（df.fillna()）或删除空列（df.dropna()）等操作。

5. 总结

通过本文的详细讲解和示例代码，您应该已经掌握了如何有效地处理USDA FDC API的分页问题，从而获取完整的营养数据。核心在于理解API的分页机制，利用pageSize和pageNumber参数，并结合迭代请求和健壮的错误处理机制。这种方法不仅适用于USDA FDC API，也适用于大多数采用分页策略的RESTful API。在实际开发中，始终优先查阅API官方文档，它是解决此类问题的最佳指南。

以上就是深入理解与实现USDA FDC API数据分页获取的详细内容，更多请关注知识资源分享宝库其它相关文章！