Asyncio Retry Utility

6.3.1 verified Fri Apr 17 auth: no python

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.

pip install aioretry

Common errors

error 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.

error 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.

error 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.

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())