How to retry async aiohttp requests depending on the status code

I am using an API and sometimes it returns some odd status codes which could be fixed by simply retrying the same request. I am using aiohttp to do submit requests to this api asynchronously.

I am also using the backoff library to retry requests, however it appears that requests are still not being retried upon 401 status responses.

 @backoff.on_exception(backoff.expo, aiohttp.ClientError, max_tries=11, max_time=60) async def get_user_timeline(self, session, user_id, count, max_id, trim_user, include_rts, tweet_mode): params = { 'user_id': user_id, 'trim_user': trim_user, 'include_rts': include_rts, 'tweet_mode': tweet_mode, 'count': count } if (max_id and max_id != -1): params.update({'max_id': max_id}) headers = { 'Authorization': 'Bearer {}'.format(self.access_token) } users_lookup_url = "/1.1/statuses/user_timeline.json" url = self.base_url + users_lookup_url async with session.get(url, params=params, headers=headers) as response: result = await response.json() response = { 'result': result, 'status': response.status, 'headers': response.headers } return response 

I would like all requests to be retired up to 10 times if the response has a status code other than 200 or 429.

4 Answers

I made a simple library, that can help you:

Code like this should solve your problem:

from aiohttp import ClientSession from aiohttp_retry import RetryClient statuses = {x for x in range(100, 600)} statuses.remove(200) statuses.remove(429) async with ClientSession() as client: retry_client = RetryClient(client) async with retry_client.get("", retry_attempts=10, retry_for_statuses=statuses) as response: text = await response.text() print(text) await retry_client.close() 

Instead google.com use your own url

By default aiohttp doesn't raise exception for non-200 status. You should change it passing raise_for_status=True (doc):

async with session.get(url, params=params, headers=headers, raise_for_status=True) as response: 

It should raise exception for any statuses 400 or higher and thus trigger backoff.

Codes 2xx shouldn't be probably retried since these aren't errors.

Anyway if you still want to raise for "other than 200 or 429" you can do it manually:

if response.status not in (200, 429,): raise aiohttp.ClientResponseError() 

Mikhail's answer cover how to raise exception for status codes 4XX and 5XX, but if you want your coroutine to be retried for these status codes, take a look at async_retrying library -

A simple example is given below.

import asyncio from aiohttp import ClientSession from async_retrying import retry @retry(attempts=2) async def hit_url(url, session: ClientSession): async with session.get(url) as response: print("Calling URL : %s", url) await response.text() return response.status async def main(): urls = [ "", "" ] api_calls = [] async with ClientSession(raise_for_status=True) as session: for url in urls: api_calls.append(hit_url(session=session, url=url)) await asyncio.gather(*api_calls, return_exceptions=False) asyncio.run(main()) 

Maybe it's too old, but for any people wondering how to build such solution

RequestData and ErrorResponseData are your custom classes, it's not something built in

class DataAPI: def __init__(self, api_data_converter: APIDataConverter): self.api_data_converter = api_data_converter async def _bound_fetch(self, request_data: RequestData, session): try: async with session.get(request_data.url, raise_for_status=True) as response: return ResponseData(await response.text()) except aiohttp.ClientConnectionError as e: Logging.log_exception('Connection error: {}'.format(str(e))) return ErrorResponseData(url=request_data.url, request_data=request_data) except Exception as e: Logging.log_exception('Data API error: {}'.format(str(e))) return ErrorResponseData(url=request_data.url, request_data=request_data) async def _run_requests(self, request_data: List[RequestData]): for rd in request_data: Logging.log_info('Request: {}'.format(rd.url)) async with aiohttp.ClientSession(timeout=ClientTimeout(total=80)) as session: tasks = [] for rd in request_data: task = asyncio.ensure_future(self._bound_fetch(rd, session)) tasks.append(task) responses = asyncio.gather(*tasks) return await responses def get_data(self, request_data: List[RequestData]): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) skipped = request_data responses: List[ResponseData] = [] for _ in range(2): # specify your retry count instead of 2 interm_responses = loop.run_until_complete(asyncio.ensure_future(self._run_requests(skipped))) skipped = [] for resp in interm_responses: if isinstance(resp, ErrorResponseData): skipped.append(resp.request_data) else: responses.append(resp) if not skipped: break if skipped: Logging.log_critical('Failed urls remaining') for resp in responses: data = self.api_data_converter.convert(resp.response) if not data: Logging.log_exception('Data API error') dt = dateutil.parser.parse(data[-1]['dt']) resp.response = data resp.last_candle_dt = dt return responses