Replace tenacity.retry with our own decorator

news/10822.vendor.rst

@@ -0,0 +1 @@
+Remove vendored tenacity.

src/pip/_internal/utils/filesystem.py

@@ -7,10 +7,9 @@
 from tempfile import NamedTemporaryFile
 from typing import Any, BinaryIO, Generator, List, Union, cast
 
-from pip._vendor.tenacity import retry, stop_after_delay, wait_fixed
-
 from pip._internal.utils.compat import get_path_uid
 from pip._internal.utils.misc import format_size
+from pip._internal.utils.retry import retry
 
 
 def check_path_owner(path: str) -> bool:
@@ -65,10 +64,7 @@ def adjacent_tmp_file(path: str, **kwargs: Any) -> Generator[BinaryIO, None, Non
             os.fsync(result.fileno())
 
 
-# Tenacity raises RetryError by default, explicitly raise the original exception
-_replace_retry = retry(reraise=True, stop=stop_after_delay(1), wait=wait_fixed(0.25))
-
-replace = _replace_retry(os.replace)
+replace = retry(stop_after_delay=1, wait=0.25)(os.replace)
 
 
 # test_writable_dir and _test_writable_dir_win are copied from Flit,

src/pip/_internal/utils/misc.py

@@ -36,12 +36,12 @@
 
 from pip._vendor.packaging.requirements import Requirement
 from pip._vendor.pyproject_hooks import BuildBackendHookCaller
-from pip._vendor.tenacity import retry, stop_after_delay, wait_fixed
 
 from pip import __version__
 from pip._internal.exceptions import CommandError, ExternallyManagedEnvironment
 from pip._internal.locations import get_major_minor_version
 from pip._internal.utils.compat import WINDOWS
+from pip._internal.utils.retry import retry
 from pip._internal.utils.virtualenv import running_under_virtualenv
 
 __all__ = [
@@ -120,8 +120,7 @@ def get_prog() -> str:
 
 
 # Retry every half second for up to 3 seconds
-# Tenacity raises RetryError by default, explicitly raise the original exception
-@retry(reraise=True, stop=stop_after_delay(3), wait=wait_fixed(0.5))
+@retry(stop_after_delay=3, wait=0.5)
 def rmtree(
     dir: str,
     ignore_errors: bool = False,

src/pip/_internal/utils/retry.py

@@ -0,0 +1,40 @@
+import functools
+from time import monotonic, 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:
+            start_time = monotonic()
+            while True:
+                try:
+                    return func(*args, **kwargs)
+                except Exception:
+                    if monotonic() - start_time > stop_after_delay:
+                        raise
+                    sleep(wait)
+
+        return retry_wrapped
+
+    return wrapper

src/pip/_internal/utils/temp_dir.py

@@ -208,7 +208,7 @@ def onerror(
 
         if self.ignore_cleanup_errors:
             try:
-                # first try with tenacity; retrying to handle ephemeral errors
+                # first try with @retry; retrying to handle ephemeral errors
                 rmtree(self._path, ignore_errors=False)
             except OSError:
                 # last pass ignore/log all errors

src/pip/_vendor/__init__.py

@@ -110,7 +110,6 @@ def vendored(modulename):
     vendored("rich.style")
     vendored("rich.text")
     vendored("rich.traceback")
-    vendored("tenacity")
     if sys.version_info < (3, 11):
         vendored("tomli")
     vendored("truststore")