如何在Python中实现单次同步请求与多次异步请求？

你可以使用httpx的异步客户端结合asyncio实现后续请求的并发执行，以下是改造后的完整示例代码：

import asyncio
import httpx

def process_data(data1, data2, data3):
    # 保留你原有的数据处理逻辑
    print("Processing data...")

async def fetch_id():
    # 同步获取ID（和原逻辑流程一致，也可改为异步，见下方补充）
    response = httpx.get('https://api.example.com/get_id')
    return response.json()['id']

async def fetch_data(session, target_id, endpoint):
    url = f'https://api.example.com/data/{target_id}/{endpoint}'
    response = await session.get(url)
    return response.json()

async def main():
    # 第一步：获取关联ID
    target_id = await fetch_id()
    
    # 第二步：创建异步客户端会话（复用连接提升性能），并发发起数据请求
    async with httpx.AsyncClient() as session:
        data1, data2, data3 = await asyncio.gather(
            fetch_data(session, target_id, 'info1'),
            fetch_data(session, target_id, 'info2'),
            fetch_data(session, target_id, 'info3')
        )
    
    # 第三步：处理返回的所有数据
    process_data(data1, data2, data3)

if __name__ == "__main__":
    asyncio.run(main())

关键逻辑说明 #

• httpx.AsyncClient：异步HTTP客户端，支持并发请求并复用连接，比同步请求更高效
• asyncio.gather()：批量执行异步任务，等待所有任务完成后返回结果，三个数据请求会同时发起，无需依次等待
• async/await：异步函数标识和异步调用关键字，确保异步操作被正确调度

补充：将初始ID请求也改为异步 #

如果希望全流程异步，只需修改fetch_id函数：

async def fetch_id():
    async with httpx.AsyncClient() as session:
        response = await session.get('https://api.example.com/get_id')
        return response.json()['id']

异常处理建议（可选） #

为避免单个请求失败导致整个流程中断，可在fetch_data中添加异常捕获：

async def fetch_data(session, target_id, endpoint):
    url = f'https://api.example.com/data/{target_id}/{endpoint}'
    try:
        response = await session.get(url)
        response.raise_for_status()  # 触发HTTP错误状态码的异常
        return response.json()
    except httpx.HTTPError as e:
        print(f"请求{endpoint}失败: {str(e)}")
        return None  # 根据业务需求返回默认值或抛出异常

内容的提问来源于stack exchange，提问作者Justin Iven Müller