Spaces:
Paused
Paused
File size: 1,391 Bytes
4efde5d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
import asyncio
from typing import TypeVar, Callable, Awaitable, Optional
T = TypeVar("T")
async def retry(
fn: Callable[[], Awaitable[T]],
max_attempts: int = 3,
delay_seconds: int = 1,
) -> T:
"""
Retry an async function with exponential backoff.
Args:
fn: The async function to retry
max_attempts: Maximum number of attempts
delay_seconds: Delay between attempts in seconds
Returns:
The result of the function call
Raises:
The last exception if all attempts fail
Example:
```python
async def fetch_data():
# Some operation that might fail
return await api_call()
try:
result = await retry(fetch_data, max_attempts=3, delay_seconds=2)
print(f"Success: {result}")
except Exception as e:
print(f"Failed after all retries: {e}")
```
"""
if max_attempts <= 0:
raise ValueError("max_attempts must be greater than zero")
last_error: Optional[Exception] = None
for attempt in range(1, max_attempts + 1):
try:
return await fn()
except Exception as error:
last_error = error
if attempt == max_attempts:
break
await asyncio.sleep(delay_seconds)
if last_error:
raise last_error
raise RuntimeError("Unexpected: last_error is None")
|