























Design a retryer function that takes an asynchronous function and a maximum number of retries. The retry delay should grow exponentially (e.g., 2 seconds, 4 seconds, 8 seconds, etc.). If the asynchronous function succeeds within the maximum retries, return the result; if it fails after the maximum retries, throw an error.
func: an asynchronous function to retry, with no parameters.max_retries: an integer specifying the maximum number of retries.async def my_async_function():
# Simulate a potentially failing async call
result = await retryer(my_async_function, 3)
await asyncio.sleep() to simulate wait time.1 import asyncio 2 from typing import Callable, Awaitable, TypeVar 3 4 T = TypeVar("T") 5 6 async def retryer( 7 func: Callable[[], Awaitable[T]], 8 max_retries: int 9 ) -> T: 10 """ 11 Retry an async function with exponential backoff. 12 13 :param func: async function with no parameters 14 :param max_retries: maximum number of retries (after first attempt) 15 :return: result of func() 16 :raises: last exception if all attempts fail 17 """ 18 if max_retries < 0: 19 raise ValueError("max_retries must be >= 0") 20 21 attempt = 0 22 delay = 2 # initial delay in seconds 23 max_delay = 16 24 25 while True: 26 try: 27 return await func() 28 except Exception as e: 29 if attempt >= max_retries: 30 raise e 31 32 await asyncio.sleep(delay) 33 34 delay = min(delay * 2, max_delay) 35 attempt += 1
or
1 async def retryer(func, max_retries): 2 delay = 2 3 max_delay = 16 4 5 for attempt in range(max_retries + 1): 6 try: 7 return await func() 8 except Exception as e: 9 if attempt == max_retries: 10 raise e 11 await asyncio.sleep(delay) 12 delay = min(delay * 2, max_delay)
Test Case
1 import random 2 import asyncio 3 4 async def my_async_function(): 5 if random.random() < 0.7: 6 raise ValueError("Random failure") 7 return "Success!" 8 9 async def main(): 10 try: 11 result = await retryer(my_async_function, 3) 12 print("Result:", result) 13 except Exception as e: 14 print("Final failure:", e) 15 16 asyncio.run(main())
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。