_asyncio.py 3.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. # Copyright 2016 Étienne Bersac
  2. # Copyright 2016 Julien Danjou
  3. # Copyright 2016 Joshua Harlow
  4. # Copyright 2013-2014 Ray Holder
  5. #
  6. # Licensed under the Apache License, Version 2.0 (the "License");
  7. # you may not use this file except in compliance with the License.
  8. # You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import functools
  18. import sys
  19. import typing
  20. from asyncio import sleep
  21. from pip._vendor.tenacity import AttemptManager
  22. from pip._vendor.tenacity import BaseRetrying
  23. from pip._vendor.tenacity import DoAttempt
  24. from pip._vendor.tenacity import DoSleep
  25. from pip._vendor.tenacity import RetryCallState
  26. WrappedFn = typing.TypeVar("WrappedFn", bound=typing.Callable)
  27. _RetValT = typing.TypeVar("_RetValT")
  28. class AsyncRetrying(BaseRetrying):
  29. def __init__(self, sleep: typing.Callable[[float], typing.Awaitable] = sleep, **kwargs: typing.Any) -> None:
  30. super().__init__(**kwargs)
  31. self.sleep = sleep
  32. async def __call__( # type: ignore # Change signature from supertype
  33. self,
  34. fn: typing.Callable[..., typing.Awaitable[_RetValT]],
  35. *args: typing.Any,
  36. **kwargs: typing.Any,
  37. ) -> _RetValT:
  38. self.begin()
  39. retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs)
  40. while True:
  41. do = self.iter(retry_state=retry_state)
  42. if isinstance(do, DoAttempt):
  43. try:
  44. result = await fn(*args, **kwargs)
  45. except BaseException: # noqa: B902
  46. retry_state.set_exception(sys.exc_info())
  47. else:
  48. retry_state.set_result(result)
  49. elif isinstance(do, DoSleep):
  50. retry_state.prepare_for_next_attempt()
  51. await self.sleep(do)
  52. else:
  53. return do
  54. def __aiter__(self) -> "AsyncRetrying":
  55. self.begin()
  56. self._retry_state = RetryCallState(self, fn=None, args=(), kwargs={})
  57. return self
  58. async def __anext__(self) -> typing.Union[AttemptManager, typing.Any]:
  59. while True:
  60. do = self.iter(retry_state=self._retry_state)
  61. if do is None:
  62. raise StopAsyncIteration
  63. elif isinstance(do, DoAttempt):
  64. return AttemptManager(retry_state=self._retry_state)
  65. elif isinstance(do, DoSleep):
  66. self._retry_state.prepare_for_next_attempt()
  67. await self.sleep(do)
  68. else:
  69. return do
  70. def wraps(self, fn: WrappedFn) -> WrappedFn:
  71. fn = super().wraps(fn)
  72. # Ensure wrapper is recognized as a coroutine function.
  73. @functools.wraps(fn)
  74. async def async_wrapped(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
  75. return await fn(*args, **kwargs)
  76. # Preserve attributes
  77. async_wrapped.retry = fn.retry
  78. async_wrapped.retry_with = fn.retry_with
  79. return async_wrapped