ask_candid/base/api_base_async.py

from typing import Any
import json

import aiohttp


class BaseAsyncAPI:

    def __init__(self, url: str, headers: dict[str, Any] | None = None, retries: int = 3) -> None:
        self.__url = url
        self.__headers = headers
        self.__retries = max(retries, 5)

    async def get(self, **request_kwargs):
        session_timeout = aiohttp.ClientTimeout(total=30)
        async with aiohttp.ClientSession(headers=self.__headers, timeout=session_timeout) as session:
            output = {}
            count = 1
            while True:
                if count >= self.__retries:
                    break
                async with session.get(url=self.__url, params=request_kwargs) as r:
                    if r.status in {429, 500, 502, 503, 504}:
                        count += 1
                    elif r.status == 200:
                        output = await r.json()
                        break
                    else:
                        break
        return output

    async def post(self, payload: dict[str, Any]):
        session_timeout = aiohttp.ClientTimeout(total=30)
        async with aiohttp.ClientSession(headers=self.__headers, timeout=session_timeout) as session:
            output = {}
            count = 1
            while True:
                if count >= self.__retries:
                    break
                async with session.post(url=self.__url, data=json.dumps(payload).encode('utf8')) as r:
                    if r.status in {429, 500, 502, 503, 504}:
                        count += 1
                    elif r.status == 200:
                        output = await r.json()
                        break
                    else:
                        break
        return output