Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tenacity/asyncio/retry.py: 41%

1# Copyright 2016–2021 Julien Danjou 

2# Copyright 2016 Joshua Harlow 

3# Copyright 2013-2014 Ray Holder 

4# 

5# Licensed under the Apache License, Version 2.0 (the "License"); 

6# you may not use this file except in compliance with the License. 

7# You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, software 

12# distributed under the License is distributed on an "AS IS" BASIS, 

13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

14# See the License for the specific language governing permissions and 

15# limitations under the License. 

16import abc 

17import typing 

18 

19from tenacity import _utils 

20from tenacity import retry_base 

21 

22if typing.TYPE_CHECKING: 

23 from tenacity import RetryCallState 

24 

25 

26class async_retry_base(retry_base): 

27 """Abstract base class for async retry strategies.""" 

28 

29 @abc.abstractmethod 

30 async def __call__(self, retry_state: "RetryCallState") -> bool: # type: ignore[override] 

31 pass 

32 

33 def __and__( # type: ignore[override] 

34 self, other: "typing.Union[retry_base, async_retry_base]" 

35 ) -> "retry_all": 

36 return retry_all(self, other) 

37 

38 def __rand__( # type: ignore[misc,override] 

39 self, other: "typing.Union[retry_base, async_retry_base]" 

40 ) -> "retry_all": 

41 return retry_all(other, self) 

42 

43 def __or__( # type: ignore[override] 

44 self, other: "typing.Union[retry_base, async_retry_base]" 

45 ) -> "retry_any": 

46 return retry_any(self, other) 

47 

48 def __ror__( # type: ignore[misc,override] 

49 self, other: "typing.Union[retry_base, async_retry_base]" 

50 ) -> "retry_any": 

51 return retry_any(other, self) 

52 

53 

54RetryBaseT = typing.Union[ 

55 async_retry_base, typing.Callable[["RetryCallState"], typing.Awaitable[bool]] 

56] 

57 

58 

59class retry_if_exception(async_retry_base): 

60 """Retry strategy that retries if an exception verifies a predicate.""" 

61 

62 def __init__( 

63 self, predicate: typing.Callable[[BaseException], typing.Awaitable[bool]] 

64 ) -> None: 

65 self.predicate = predicate 

66 

67 async def __call__(self, retry_state: "RetryCallState") -> bool: # type: ignore[override] 

68 if retry_state.outcome is None: 

69 raise RuntimeError("__call__() called before outcome was set") 

70 

71 if retry_state.outcome.failed: 

72 exception = retry_state.outcome.exception() 

73 if exception is None: 

74 raise RuntimeError("outcome failed but the exception is None") 

75 return await self.predicate(exception) 

76 else: 

77 return False 

78 

79 

80class retry_if_result(async_retry_base): 

81 """Retries if the result verifies a predicate.""" 

82 

83 def __init__( 

84 self, predicate: typing.Callable[[typing.Any], typing.Awaitable[bool]] 

85 ) -> None: 

86 self.predicate = predicate 

87 

88 async def __call__(self, retry_state: "RetryCallState") -> bool: # type: ignore[override] 

89 if retry_state.outcome is None: 

90 raise RuntimeError("__call__() called before outcome was set") 

91 

92 if not retry_state.outcome.failed: 

93 return await self.predicate(retry_state.outcome.result()) 

94 else: 

95 return False 

96 

97 

98class retry_any(async_retry_base): 

99 """Retries if any of the retries condition is valid.""" 

100 

101 def __init__(self, *retries: typing.Union[retry_base, async_retry_base]) -> None: 

102 self.retries = retries 

103 

104 async def __call__(self, retry_state: "RetryCallState") -> bool: # type: ignore[override] 

105 result = False 

106 for r in self.retries: 

107 result = result or await _utils.wrap_to_async_func(r)(retry_state) 

108 if result: 

109 break 

110 return result 

111 

112 

113class retry_all(async_retry_base): 

114 """Retries if all the retries condition are valid.""" 

115 

116 def __init__(self, *retries: typing.Union[retry_base, async_retry_base]) -> None: 

117 self.retries = retries 

118 

119 async def __call__(self, retry_state: "RetryCallState") -> bool: # type: ignore[override] 

120 result = True 

121 for r in self.retries: 

122 result = result and await _utils.wrap_to_async_func(r)(retry_state) 

123 if not result: 

124 break 

125 return result