From 4aab4087dc97906d0b9890035401175cdaab32d4 Mon Sep 17 00:00:00 2001 From: blackhao <13851610112@163.com> Date: Fri, 22 Aug 2025 02:51:50 -0500 Subject: 2.0 --- .../site-packages/pip/_internal/utils/retry.py | 45 ++++++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 .venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py (limited to '.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py') 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 0000000..27d3b6e --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pip/_internal/utils/retry.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import functools +from time import perf_counter, sleep +from typing import TYPE_CHECKING, Callable, TypeVar + +if TYPE_CHECKING: + from 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 -- cgit v1.2.3