我编写的Python多API端点请求代码是否满足并发需求？

一、现有代码的问题 #

1. 完全未实现并发请求 #

你当前的代码是串行执行的：必须等/products的请求完全拿到响应后，才会发送/users的请求，两个请求没有同时执行，完全没用到并发逻辑。

2. 存在潜在报错风险 #

• 如果API返回的产品/用户数据是空列表，products_data[0].keys()或users_data[0].keys()会直接抛出IndexError，因为空列表没有第0个元素。
• 用户数据里包含嵌套结构（比如address是字典），直接写入user.values()会把整个字典以字符串形式存入CSV，后续读取时无法正常解析。

3. 基础功能可用但鲁棒性不足 #

在API正常返回非空、无嵌套数据的场景下，代码能成功将数据写入对应CSV文件，但只要出现异常情况就会崩溃。

二、实现并发请求的修正代码 #

用Python标准库的concurrent.futures.ThreadPoolExecutor可轻松实现并发，同时修复上述潜在问题：

import requests
import csv
from concurrent.futures import ThreadPoolExecutor

def fetch_and_save(url, filename):
    try:
        response = requests.get(url)
        response.raise_for_status()  # 自动捕获4xx/5xx状态码错误
        data = response.json()
        
        if not data:
            print(f"警告：{filename}对应的API返回空数据")
            return
        
        # 展开嵌套字段，例如将address.city转为address_city形式
        flattened_data = []
        for item in data:
            flattened_item = {}
            for key, value in item.items():
                if isinstance(value, dict):
                    for sub_key, sub_val in value.items():
                        flattened_item[f"{key}_{sub_key}"] = sub_val
                else:
                    flattened_item[key] = value
            flattened_data.append(flattened_item)
        
        # 使用DictWriter更安全地写入CSV
        with open(filename, "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=flattened_data[0].keys())
            writer.writeheader()
            writer.writerows(flattened_data)
        print(f"{filename} 保存成功")
    except Exception as e:
        print(f"处理{filename}时出错: {str(e)}")

# 定义任务列表
tasks = [
    ("https://fakestoreapi.com/products", "products.csv"),
    ("https://fakestoreapi.com/users", "users.csv")
]

# 启动线程池并发执行任务
with ThreadPoolExecutor(max_workers=2) as executor:
    for url, filename in tasks:
        executor.submit(fetch_and_save, url, filename)

三、修正点说明 #

• 并发实现：通过线程池同时启动两个任务，两个API请求会同步发送，无需等待一个请求完成再执行另一个，能有效减少总耗时。
• 鲁棒性优化：
  • 用response.raise_for_status()自动处理请求失败的情况，无需手动判断状态码。
  • 增加空数据判断，避免空列表索引报错。
  • 将嵌套字段展开为一级字段，保证CSV数据的可读性与可解析性。
  • 使用csv.DictWriter处理写入操作，无需手动维护键值顺序，降低出错概率。

内容的提问来源于stack exchange，提问作者Ömer Talha Demirci