venv/lib/python3.11/site-packages/pip/_vendor/tenacity/tornadoweb.py

   1 # Copyright 2017 Elisey Zanko
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import sys
  16 import typing
  17
  18 from pip._vendor.tenacity import BaseRetrying
  19 from pip._vendor.tenacity import DoAttempt
  20 from pip._vendor.tenacity import DoSleep
  21 from pip._vendor.tenacity import RetryCallState
  22
  23 from tornado import gen
  24
  25 if typing.TYPE_CHECKING:
  26     from tornado.concurrent import Future
  27
  28 _RetValT = typing.TypeVar("_RetValT")
  29
  30
  31 class TornadoRetrying(BaseRetrying):
  32     def __init__(self, sleep: "typing.Callable[[float], Future[None]]" = gen.sleep, **kwargs: typing.Any) -> None:
  33         super().__init__(**kwargs)
  34         self.sleep = sleep
  35
  36     @gen.coroutine  # type: ignore[misc]
  37     def __call__(
  38         self,
  39         fn: "typing.Callable[..., typing.Union[typing.Generator[typing.Any, typing.Any, _RetValT], Future[_RetValT]]]",
  40         *args: typing.Any,
  41         **kwargs: typing.Any,
  42     ) -> "typing.Generator[typing.Any, typing.Any, _RetValT]":
  43         self.begin()
  44
  45         retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs)
  46         while True:
  47             do = self.iter(retry_state=retry_state)
  48             if isinstance(do, DoAttempt):
  49                 try:
  50                     result = yield fn(*args, **kwargs)
  51                 except BaseException:  # noqa: B902
  52                     retry_state.set_exception(sys.exc_info())  # type: ignore[arg-type]
  53                 else:
  54                     retry_state.set_result(result)
  55             elif isinstance(do, DoSleep):
  56                 retry_state.prepare_for_next_attempt()
  57                 yield self.sleep(do)
  58             else:
  59                 raise gen.Return(do)