如何修改Python asyncio重试逻辑实现200响应下按指定结果重试

核心思路 #

• async_retrying 库的重试触发条件是被装饰的异步函数抛出异常，原有逻辑仅非200状态码时raise_for_status=True会抛出HTTP异常触发重试
• 要实现200状态码下的业务逻辑重试，只需要在判断到响应存在指定key时，主动抛出异常即可触发重试机制

调整后的完整客户端代码 #

# load required libraries
import json
import asyncio
import aiohttp
from async_retrying import retry

base_url = "http://localhost:1050/hello?rid="

# 自定义业务重试异常，方便区分重试触发原因
class BusinessRetryException(Exception):
    pass

# async ginger call 新增url入参，修复原代码全局变量竞态问题
@retry(attempts=3)
async def async_ginger_call(url):
    connector = aiohttp.TCPConnector(limit=3)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.post(url, raise_for_status=True, timeout=300) as response:
            result = await response.text()
            result_json = json.loads(result)
            # 业务重试判断：此处以检查Result下是否存在Warnings字段为例，可按需调整判断规则
            if "Warnings" in result_json.get("Result", {}):
                # 主动抛出异常触发重试
                raise BusinessRetryException("响应包含指定重试key: Warnings")
            return result_json

reqs = 2
tasks = []
connector = aiohttp.TCPConnector(limit=reqs)
async with aiohttp.ClientSession(connector=connector) as session:
    for i in range(reqs):
        url = base_url + str(i)
        # 传入url参数
        tasks.append(async_ginger_call(url))

    results = await asyncio.gather(*tasks, return_exceptions=True)

验证说明 #

你当前提供的Flask测试服务默认返回的200响应本身就携带Warnings字段，运行修改后的代码即可观察到重试逻辑被触发。如果需要测试正常返回场景，修改Flask服务的valid_response去掉Warnings字段即可。

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