Run the loop body until "break" stops the loop. After *timeout* seconds, raise an AssertionError if *error* is true, or just stop if *error is false. Example: for _ in support.busy_retry(support.SHORT_TIMEOUT): if check(): break Example of
(timeout, err_msg=None, /, *, error=True)
| 2680 | |
| 2681 | |
| 2682 | def busy_retry(timeout, err_msg=None, /, *, error=True): |
| 2683 | """ |
| 2684 | Run the loop body until "break" stops the loop. |
| 2685 | |
| 2686 | After *timeout* seconds, raise an AssertionError if *error* is true, |
| 2687 | or just stop if *error is false. |
| 2688 | |
| 2689 | Example: |
| 2690 | |
| 2691 | for _ in support.busy_retry(support.SHORT_TIMEOUT): |
| 2692 | if check(): |
| 2693 | break |
| 2694 | |
| 2695 | Example of error=False usage: |
| 2696 | |
| 2697 | for _ in support.busy_retry(support.SHORT_TIMEOUT, error=False): |
| 2698 | if check(): |
| 2699 | break |
| 2700 | else: |
| 2701 | raise RuntimeError('my custom error') |
| 2702 | |
| 2703 | """ |
| 2704 | if timeout <= 0: |
| 2705 | raise ValueError("timeout must be greater than zero") |
| 2706 | |
| 2707 | start_time = time.monotonic() |
| 2708 | deadline = start_time + timeout |
| 2709 | |
| 2710 | while True: |
| 2711 | yield |
| 2712 | |
| 2713 | if time.monotonic() >= deadline: |
| 2714 | break |
| 2715 | |
| 2716 | if error: |
| 2717 | dt = time.monotonic() - start_time |
| 2718 | msg = f"timeout ({dt:.1f} seconds)" |
| 2719 | if err_msg: |
| 2720 | msg = f"{msg}: {err_msg}" |
| 2721 | raise AssertionError(msg) |
| 2722 | |
| 2723 | |
| 2724 | def sleeping_retry(timeout, err_msg=None, /, |
no outgoing calls
searching dependent graphs…