Retry repeats a failed or timed-out operation when the failure may be transient, such as a networking hiccup or temporary load-balancer problem. Waiting briefly can recover useful results, but may also make users wait longer for the same failure. section 10.3.4
A reusable asynchronous retry operation:
- Accepts a callable returning an awaitable, a retry limit, an operation timeout, and an interval between attempts. section 10.3.4
- Wraps each attempt with asyncio.wait_for, catches failures or timeouts, logs the error, and sleeps before retrying. section 10.3.4
- Returns successful results or raises a dedicated TooManyRetries exception after the allowed attempts. section 10.3.4
Testing failure and timeout paths
A coroutine that always raises and one that always exceeds its timeout both exhaust three attempts and produce TooManyRetries, while logging each retry.
Listing 10.9 A retry coroutine
import asyncio
import logging
from typing import Callable, Awaitable
class TooManyRetries(Exception):
pass
async def retry(coro: Callable[[], Awaitable],
max_retries: int,
timeout: float,
retry_interval: float):
for retry_num in range(0, max_retries):
try:
return await asyncio.wait_for(coro(), timeout=timeout) #1
except Exception as e:
logging.exception(f'Exception while waiting (tried {retry_num} times), retrying.', exc_info=e) #2
await asyncio.sleep(retry_interval)
raise TooManyRetries() #3
View the complete asyncio retry coroutine, including timeout enforcement, logging, delay, and exhaustion handling. section 10.3.4
