How can I implement retry policy with httpx in Python?

Question

I need to communicate with other services in my Python and FastAPI application, therefore I use the httpx library to be able to communicate asynchronously. So, I have the following code for POST requests:

from typing import Any, Dict, Optional, Tuple

from fastapi import File
from httpx._client import AsyncClient


async def post(
    *,
    url: str,
    files: Optional[Dict[str, File]] = None,
    json: Optional[Dict[str, Any]] = None,
    data: Optional[Dict[str, str]] = None,
    params: Optional[Dict[str, str]] = None,
    timeout: int = 10000
) -> Tuple[bool, Any]:
    try:
        async with AsyncClient() as client:
            response = await client.post(url, files=files, json=json, data=data, timeout=timeout)
            response = response.json() if response.status_code == 200 else None
            if not response:
                return False, None
            return True, response
    except Exception as e:
        print(e)
        return False, None

I would like to implement a retry policy so that if a request fails, it is retried, for example, up to 3 times. Is this possible and makes sense with httpx and async? I was looking at some tutorials on the internet but they seem to be outdated since the information they contain does not work

Update: I tried the following approach with HTTPTransport but it didn't work for me:

from httpx import HTTPTransport # here

 try:
        async with AsyncClient(transport=transport) as client: # here
            response = await client.post(url, files=files, json=json, data=data, timeout=timeout)
            response = response.json() if response.status_code == 200 else None
            if not response:
                return False, None
            return True, response
    except Exception as e:
        print(e)
        return False, None

transport = HTTPTransport(retries=3)

I get: 'HTTPTransport' object has no attribute 'aenter'

Answer 1

With the Async HTTPTransport class of httpx you can configure retry:

from typing import Any, Dict, Optional, Tuple

from fastapi import File
from httpx import AsyncHTTPTransport
from httpx._client import AsyncClient

transport = AsyncHTTPTransport(retries=3)

async def post(
    *,
    url: str,
    files: Optional[Dict[str, File]] = None,
    json: Optional[Dict[str, Any]] = None,
    data: Optional[Dict[str, str]] = None,
    params: Optional[Dict[str, str]] = None,
    timeout: int = 100
) -> Tuple[bool, Any]:
    if params:
        parameters = [key + "=" + parameter for key, parameter in params.items()]
        parameters = "&".join(parameters)
        url += "?" + parameters
    try:
        async with AsyncClient(transport=transport) as client:
            response = await client.post(url, files=files, json=json, data=data, timeout=timeout)
            response = response.json() if response.status_code == 200 else None
            if not response:
                return False, None
            return True, response
    except Exception as e:
        print(e)
        return False, None

I tried the above and it worked!