Python Asyncio：基于首次请求响应发起二次请求实现方案

首先，你需要修正原代码的几个基础问题：比如aiottp应为aiohttp，任务创建时的参数传递错误，以及核心的「首次请求成功后立即触发后续操作」逻辑——不能用asyncio.gather()等待全部任务完成，而是要逐个监控任务的完成状态，一旦有任务成功就立刻执行后续API调用。

实现思路 #

• 为每个请求封装独立的异步函数，创建任务集合
• 使用asyncio.wait()的FIRST_COMPLETED模式，每次只等待第一个完成的任务
• 检查任务结果，若首次成功则立即调用目标API
• 持续监控剩余任务，记录所有成功的请求

修正后的完整代码 #

import asyncio
import aiohttp
import json  # 用于安全解析payload字符串

async def single_request(session, url, payload_str):
    """单个请求的异步逻辑，返回请求是否成功及响应内容"""
    try:
        # 把字符串格式的payload转成字典（比eval更安全）
        payload = json.loads(payload_str)
        async with session.get(url, params=payload) as response:
            if response.status == 200:
                # 记录成功请求（可替换为写入日志/数据库）
                print(f"请求成功，参数：{payload_str}")
                return True, await response.text()
            return False, await response.text()
    except Exception as e:
        print(f"请求失败：{str(e)}")
        return False, str(e)

async def call_another_api(session):
    """首次请求成功后需要调用的另一API"""
    another_api_url = "bar"
    async with session.get(another_api_url) as resp:
        print(f"后续API调用完成，状态码：{resp.status}")
        return await resp.text()

async def main():
    target_url = "foo"
    payload_list = ['{"a": 1}', '{"b": 2}']
    first_success_triggered = False  # 标记是否已触发后续API

    async with aiohttp.ClientSession() as session:
        # 创建所有请求任务，存入集合
        tasks = {asyncio.create_task(single_request(session, target_url, p)) for p in payload_list}
        
        # 循环处理完成的任务，直到所有任务结束
        while tasks:
            # 等待第一个完成的任务，不阻塞其他请求
            done_tasks, pending_tasks = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            
            for task in done_tasks:
                is_success, result = await task
                # 首次成功时触发后续API
                if is_success and not first_success_triggered:
                    await call_another_api(session)
                    first_success_triggered = True
                # 这里可以统一处理所有成功/失败的请求记录
            
            # 更新任务集合，移除已完成的任务
            tasks = pending_tasks

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

关键细节说明 #

• asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)是核心：它会在第一个任务完成时立即返回，不用等全部请求结束
• 用first_success_triggered标记避免重复调用后续API
• 用json.loads()解析字符串payload比eval()更安全，避免恶意代码执行风险

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