Asyncio Retry Utility

6.3.1 · active · verified Fri Apr 17

aioretry is an asyncio utility for Python 3.7+ that provides flexible retry mechanisms for asynchronous operations. It supports various retry policies, including fixed delay, exponential backoff, and custom strategies, handling transient failures in async code gracefully. The current version is 6.3.1, and it maintains an 'active' release cadence with regular updates.

Common errors

```
AttributeError: 'float' object has no attribute 'total_seconds'
```
cause Attempting to call `.total_seconds()` on `RetryInfo.since` after `aioretry` 6.0.0. `since` is now a `float` (monotonic time), not a `datetime` object.

fix Replace `(datetime.now() - info.since).total_seconds()` or similar with `time.monotonic() - info.since` to calculate elapsed time, as `info.since` is already a `float`.
```
TypeError: object NoneType can't be used in 'await' expression OR TypeError: 'function' object is not awaitable
```
cause This typically occurs when trying to use an `async def` retry policy function with `aioretry` versions older than 6.2.0, or if your policy function is incorrectly defined as `async` when it should be sync (or vice versa in specific contexts).

fix If you intend to use an `async def` retry policy, upgrade `aioretry` to version 6.2.0 or newer. If your policy doesn't require `await`, define it as a regular `def` function.
```
ValueError: not enough values to unpack (expected 2, got 1) OR TypeError: 'bool' object is not iterable
```
cause A custom retry policy function returned a single boolean or an incorrect structure instead of the required `(should_give_up: bool, delay_in_seconds: float)` tuple.

fix Modify your custom retry policy to explicitly return a 2-element tuple: the first element as a boolean (`True` to give up, `False` to retry), and the second as a float for the delay in seconds. Example: `return info.fails > MAX_ATTEMPTS, info.fails * 0.1`.

Warnings

breaking The type of `RetryInfo::since` changed from `datetime` to `float` (from `time.monotonic()`) in version 6.0.0. This improves precision for measuring intervals but breaks code expecting `datetime` methods.
Fix: Update your retry policies to use `info.since` as a float directly (e.g., `time.monotonic() - info.since`) instead of `datetime` methods like `.total_seconds()`.
gotcha Asynchronous retry policies (e.g., `async def custom_policy(...)`) were introduced in version 6.2.0. If you define an async policy, ensure your `aioretry` version is 6.2.0 or newer.
Fix: Upgrade `aioretry` to `6.2.0` or higher if you need to use `async def` for your retry policies. For older versions, policies must be synchronous functions.
gotcha Custom retry policies must return a tuple `(should_give_up: bool, delay_in_seconds: float)`. Returning fewer or more values, or values of incorrect types, will lead to runtime errors.
Fix: Ensure your custom policy function (sync or async) consistently returns a tuple of exactly two elements: a boolean indicating whether to stop retrying, and a float for the delay before the next attempt.

Install

pip install aioretry Install latest version

Imports

retry
```
from aioretry import retry
```
RetryInfo
```
from aioretry import RetryInfo
```

RetryPolicyStrategy

from aioretry import RetryPolicyStrategy

Quickstart

This quickstart demonstrates using the `@retry` decorator with a custom asynchronous retry policy. The `flaky_async_call` function simulates transient failures, and `my_retry_policy` specifies when to give up and how long to delay between attempts.

import asyncio
from aioretry import retry, RetryInfo, RetryPolicyStrategy

fail_count = 0

async def my_retry_policy(info: RetryInfo) -> RetryPolicyStrategy:
    """Retry up to 3 times, with increasing delay."""
    # (should_give_up, delay_in_seconds)
    return info.fails >= 3, info.fails * 0.5

@retry(policy=my_retry_policy)
async def flaky_async_call():
    global fail_count
    fail_count += 1
    print(f"Attempt {fail_count}: Making a flaky call...")
    if fail_count < 3: # Succeed on the 3rd attempt
        raise ConnectionError("Simulated network issue")
    print(f"Attempt {fail_count}: Call succeeded!")
    return "Data Fetched"

async def main():
    print("Starting main program...")
    try:
        result = await flaky_async_call()
        print(f"Final Result: {result}")
    except Exception as e:
        print(f"Operation failed after retries: {e}")
    print("Program finished.")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →