如何在Python的httpx异步请求中实现重试策略？解决相关报错

你遇到的错误是因为HTTPTransport是底层传输实现，不能直接作为异步上下文管理器使用。要给httpx异步请求添加重试策略，正确的做法是通过AsyncClient的retry参数配合httpx.Retry类来配置，而不是直接在HTTPTransport上设置retries。

具体实现步骤： #

1. 导入必要的模块：

import httpx
from httpx import AsyncClient, Retry
from fastapi import FastAPI, Depends

2. 配置重试策略：
   创建Retry实例，指定最大重试次数、需要重试的HTTP状态码、允许重试的请求方法（POST请求如果是幂等操作才建议重试，需根据业务场景调整）：

retry_strategy = Retry(
    max_retries=3,
    # 针对5xx服务器错误和429限流错误重试
    status_codes=[500, 502, 503, 504, 429],
    # 允许重试POST方法（若POST接口是非幂等的，比如创建唯一资源，不建议开启）
    allowed_methods=["POST"]
)

3. 创建带重试的AsyncClient：
   将重试策略传入AsyncClient，如果需要配置传输层参数（如连接池、超时），再把HTTPTransport传给transport参数：

async def get_retry_client():
    transport = httpx.HTTPTransport(
        max_connections=100,
        keepalive_expiry=30
    )
    async with AsyncClient(transport=transport, retry=retry_strategy) as client:
        yield client

4. 在FastAPI接口中使用：
   通过依赖注入复用带重试逻辑的客户端：

app = FastAPI()

@app.post("/invoke-external")
async def invoke_external_service(client: AsyncClient = Depends(get_retry_client)):
    response = await client.post(
        "https://target-service.com/api/action",
        json={"payload": "your-data"}
    )
    response.raise_for_status()
    return response.json()

简化版（无需自定义传输层） #

如果不需要配置传输层的额外参数，直接给AsyncClient传入retry参数即可：

async with AsyncClient(retry=Retry(max_retries=3)) as client:
    response = await client.post("https://target-service.com/api/action", json={"key": "value"})

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