(
f: Callable[[], T],
max_attempts: int = 60,
sleep_secs: int = 1,
exception_types: Sequence[type[Exception]] = [AssertionError],
)
| 478 | |
| 479 | |
| 480 | def retry( |
| 481 | f: Callable[[], T], |
| 482 | max_attempts: int = 60, |
| 483 | sleep_secs: int = 1, |
| 484 | exception_types: Sequence[type[Exception]] = [AssertionError], |
| 485 | ) -> T | None: |
| 486 | result = None |
| 487 | for attempt in range(max_attempts): |
| 488 | try: |
| 489 | result = f() |
| 490 | break |
| 491 | except tuple(exception_types): |
| 492 | if attempt == max_attempts - 1: |
| 493 | raise |
| 494 | sleep(sleep_secs) |
| 495 | return result |
| 496 | |
| 497 | |
| 498 | if __name__ == "__main__": |
no test coverage detected