aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py
diff options
context:
space:
mode:
authorS. Solomon Darnell2025-03-28 21:52:21 -0500
committerS. Solomon Darnell2025-03-28 21:52:21 -0500
commit4a52a71956a8d46fcb7294ac71734504bb09bcc2 (patch)
treeee3dc5af3b6313e921cd920906356f5d4febc4ed /.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py
parentcc961e04ba734dd72309fb548a2f97d67d578813 (diff)
downloadgn-ai-master.tar.gz
two version of R2R are hereHEADmaster
Diffstat (limited to '.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py')
-rw-r--r--.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py42
1 files changed, 42 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py b/.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py
new file mode 100644
index 00000000..abfe0728
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py
@@ -0,0 +1,42 @@
+import functools
+from time import perf_counter, sleep
+from typing import Callable, TypeVar
+
+from pip._vendor.typing_extensions import ParamSpec
+
+T = TypeVar("T")
+P = ParamSpec("P")
+
+
+def retry(
+ wait: float, stop_after_delay: float
+) -> Callable[[Callable[P, T]], Callable[P, T]]:
+ """Decorator to automatically retry a function on error.
+
+ If the function raises, the function is recalled with the same arguments
+ until it returns or the time limit is reached. When the time limit is
+ surpassed, the last exception raised is reraised.
+
+ :param wait: The time to wait after an error before retrying, in seconds.
+ :param stop_after_delay: The time limit after which retries will cease,
+ in seconds.
+ """
+
+ def wrapper(func: Callable[P, T]) -> Callable[P, T]:
+
+ @functools.wraps(func)
+ def retry_wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
+ # The performance counter is monotonic on all platforms we care
+ # about and has much better resolution than time.monotonic().
+ start_time = perf_counter()
+ while True:
+ try:
+ return func(*args, **kwargs)
+ except Exception:
+ if perf_counter() - start_time > stop_after_delay:
+ raise
+ sleep(wait)
+
+ return retry_wrapped
+
+ return wrapper