Skip to main content

Simple retry client for aiohttp

Project description

Simple aiohttp retry client


Python 3.7 or higher.

Install: pip install aiohttp-retry.

"Buy Me A Coffee"

Breaking API changes

Examples of usage:

from aiohttp_retry import RetryClient, ExponentialRetry

async def main():
    retry_options = ExponentialRetry(attempts=1)
    retry_client = RetryClient(raise_for_status=False, retry_options=retry_options)
    async with retry_client.get('') as response:
    await retry_client.close()
from aiohttp import ClientSession
from aiohttp_retry import RetryClient 

async def main():
    client_session = ClientSession()
    retry_client = RetryClient(client_session=client_session)
    async with retry_client.get('') as response:

    await client_session.close()
from aiohttp_retry import RetryClient, RandomRetry

async def main():
    retry_options = RandomRetry(attempts=1)
    retry_client = RetryClient(raise_for_status=False, retry_options=retry_options)

    response = await retry_client.get('/ping')
    await retry_client.close()
from aiohttp_retry import RetryClient

async def main():
    async with RetryClient() as client:
        async with client.get('') as response:

You can change parameters between attempts by passing multiple requests params:

from aiohttp_retry import RetryClient, RequestParams, ExponentialRetry

async def main():
    retry_client = RetryClient(raise_for_status=False)

    async with retry_client.requests(
                headers={'some_header': 'some_value'},
    ) as response:
    await retry_client.close()

You can also add some logic, F.E. logging, on failures by using trace mechanic.

import logging
import sys
from types import SimpleNamespace

from aiohttp import ClientSession, TraceConfig, TraceRequestStartParams

from aiohttp_retry import RetryClient, ExponentialRetry

handler = logging.StreamHandler(sys.stdout)
logger = logging.getLogger(__name__)
retry_options = ExponentialRetry(attempts=2)

async def on_request_start(
    session: ClientSession,
    trace_config_ctx: SimpleNamespace,
    params: TraceRequestStartParams,
) -> None:
    current_attempt = trace_config_ctx.trace_request_ctx['current_attempt']
    if retry_options.attempts <= current_attempt:
        logger.warning('Wow! We are in last attempt')

async def main():
    trace_config = TraceConfig()
    retry_client = RetryClient(retry_options=retry_options, trace_configs=[trace_config])

    response = await retry_client.get('', ssl=False)

    await retry_client.close()

Look tests for more examples.
Be aware: last request returns as it is.
If the last request ended with exception, that this exception will be raised from RetryClient request


RetryClient takes the same arguments as ClientSession[docs]
RetryClient has methods:

  • request
  • get
  • options
  • head
  • post
  • put
  • patch
  • put
  • delete

They are same as for ClientSession, but take one possible additional argument:

class RetryOptionsBase:
    def __init__(
        attempts: int = 3,  # How many times we should retry
        statuses: Optional[Iterable[int]] = None,  # On which statuses we should retry
        exceptions: Optional[Iterable[Type[Exception]]] = None,  # On which exceptions we should retry
        retry_all_server_errors: bool = True,  # If should retry all 500 errors or not
        # a callback that will run on response to decide if retry
        evaluate_response_callback: Optional[EvaluateResponseCallbackType] = None,

    def get_timeout(self, attempt: int, response: Optional[Response] = None) -> float:
        raise NotImplementedError

You can specify RetryOptions both for RetryClient and it's methods. RetryOptions in methods override RetryOptions defined in RetryClient constructor.

Important: by default all 5xx responses are retried + statuses you specified as statuses param If you will pass retry_all_server_errors=False than you can manually set what 5xx errors to retry.

You can define your own timeouts logic or use:

  • ExponentialRetry with exponential backoff
  • RandomRetry for random backoff
  • ListRetry with backoff you predefine by list
  • FibonacciRetry with backoff that looks like fibonacci sequence
  • JitterRetry exponential retry with a bit of randomness

Important: you can proceed server response as an parameter for calculating next timeout.
However this response can be None, server didn't make a response or you have set up raise_for_status=True Look here for an example:

Additionally, you can specify evaluate_response_callback. It receive a ClientResponse and decide to retry or not by returning a bool. It can be useful, if server API sometimes response with malformed data.

Request Trace Context

RetryClient add current attempt number to request_trace_ctx (see examples, for more info see aiohttp doc).

Change parameters between retries

RetryClient also has a method called requests. This method should be used if you want to make requests with different params.

class RequestParams:
    method: str
    url: _RAW_URL_TYPE
    trace_request_ctx: Optional[Dict[str, Any]] = None
    kwargs: Optional[Dict[str, Any]] = None
def requests(
    params_list: List[RequestParams],
    retry_options: Optional[RetryOptionsBase] = None,
    raise_for_status: Optional[bool] = None,
) -> _RequestContext:

You can find an example of usage above or in tests.
But basically RequestParams is a structure to define params for ClientSession.request func.
method, url, headers trace_request_ctx defined outside kwargs, because they are popular.

There is also an old way to change URL between retries by specifying url as list of urls. Example:

from aiohttp_retry import RetryClient

retry_client = RetryClient()
async with retry_client.get(url=['/internal_error', '/ping']) as response:
    text = await response.text()
    assert response.status == 200
    assert text == 'Ok!'

await retry_client.close()

In this example we request /interval_error, fail and then successfully request /ping. If you specify less urls than attempts number in RetryOptions, RetryClient will request last url at last attempts. This means that in example above we would request /ping once again in case of failure.


aiohttp_retry is a typed project. It should be fully compatablie with mypy.

It also introduce one special type:

ClientType = Union[ClientSession, RetryClient]

This type can be imported by from aiohttp_retry.types import ClientType

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiohttp_retry-2.8.3.tar.gz (10.6 kB view hashes)

Uploaded source

Built Distribution

aiohttp_retry-2.8.3-py3-none-any.whl (9.8 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page