Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/tenacity/retry.py: 39%

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. 

16 

17import abc 

18import re 

19import typing 

20 

21if typing.TYPE_CHECKING: 

22 from pip._vendor.tenacity import RetryCallState 

23 

24 

25class retry_base(abc.ABC): 

26 """Abstract base class for retry strategies.""" 

27 

28 @abc.abstractmethod 

29 def __call__(self, retry_state: "RetryCallState") -> bool: 

30 pass 

31 

32 def __and__(self, other: "retry_base") -> "retry_all": 

33 return retry_all(self, other) 

34 

35 def __or__(self, other: "retry_base") -> "retry_any": 

36 return retry_any(self, other) 

37 

38 

39RetryBaseT = typing.Union[retry_base, typing.Callable[["RetryCallState"], bool]] 

40 

41 

42class _retry_never(retry_base): 

43 """Retry strategy that never rejects any result.""" 

44 

45 def __call__(self, retry_state: "RetryCallState") -> bool: 

46 return False 

47 

48 

49retry_never = _retry_never() 

50 

51 

52class _retry_always(retry_base): 

53 """Retry strategy that always rejects any result.""" 

54 

55 def __call__(self, retry_state: "RetryCallState") -> bool: 

56 return True 

57 

58 

59retry_always = _retry_always() 

60 

61 

62class retry_if_exception(retry_base): 

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

64 

65 def __init__(self, predicate: typing.Callable[[BaseException], bool]) -> None: 

66 self.predicate = predicate 

67 

68 def __call__(self, retry_state: "RetryCallState") -> bool: 

69 if retry_state.outcome is None: 

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

71 

72 if retry_state.outcome.failed: 

73 exception = retry_state.outcome.exception() 

74 if exception is None: 

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

76 return self.predicate(exception) 

77 else: 

78 return False 

79 

80 

81class retry_if_exception_type(retry_if_exception): 

82 """Retries if an exception has been raised of one or more types.""" 

83 

84 def __init__( 

85 self, 

86 exception_types: typing.Union[ 

87 typing.Type[BaseException], 

88 typing.Tuple[typing.Type[BaseException], ...], 

89 ] = Exception, 

90 ) -> None: 

91 self.exception_types = exception_types 

92 super().__init__(lambda e: isinstance(e, exception_types)) 

93 

94 

95class retry_if_not_exception_type(retry_if_exception): 

96 """Retries except an exception has been raised of one or more types.""" 

97 

98 def __init__( 

99 self, 

100 exception_types: typing.Union[ 

101 typing.Type[BaseException], 

102 typing.Tuple[typing.Type[BaseException], ...], 

103 ] = Exception, 

104 ) -> None: 

105 self.exception_types = exception_types 

106 super().__init__(lambda e: not isinstance(e, exception_types)) 

107 

108 

109class retry_unless_exception_type(retry_if_exception): 

110 """Retries until an exception is raised of one or more types.""" 

111 

112 def __init__( 

113 self, 

114 exception_types: typing.Union[ 

115 typing.Type[BaseException], 

116 typing.Tuple[typing.Type[BaseException], ...], 

117 ] = Exception, 

118 ) -> None: 

119 self.exception_types = exception_types 

120 super().__init__(lambda e: not isinstance(e, exception_types)) 

121 

122 def __call__(self, retry_state: "RetryCallState") -> bool: 

123 if retry_state.outcome is None: 

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

125 

126 # always retry if no exception was raised 

127 if not retry_state.outcome.failed: 

128 return True 

129 

130 exception = retry_state.outcome.exception() 

131 if exception is None: 

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

133 return self.predicate(exception) 

134 

135 

136class retry_if_exception_cause_type(retry_base): 

137 """Retries if any of the causes of the raised exception is of one or more types. 

138 

139 The check on the type of the cause of the exception is done recursively (until finding 

140 an exception in the chain that has no `__cause__`) 

141 """ 

142 

143 def __init__( 

144 self, 

145 exception_types: typing.Union[ 

146 typing.Type[BaseException], 

147 typing.Tuple[typing.Type[BaseException], ...], 

148 ] = Exception, 

149 ) -> None: 

150 self.exception_cause_types = exception_types 

151 

152 def __call__(self, retry_state: "RetryCallState") -> bool: 

153 if retry_state.outcome is None: 

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

155 

156 if retry_state.outcome.failed: 

157 exc = retry_state.outcome.exception() 

158 while exc is not None: 

159 if isinstance(exc.__cause__, self.exception_cause_types): 

160 return True 

161 exc = exc.__cause__ 

162 

163 return False 

164 

165 

166class retry_if_result(retry_base): 

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

168 

169 def __init__(self, predicate: typing.Callable[[typing.Any], bool]) -> None: 

170 self.predicate = predicate 

171 

172 def __call__(self, retry_state: "RetryCallState") -> bool: 

173 if retry_state.outcome is None: 

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

175 

176 if not retry_state.outcome.failed: 

177 return self.predicate(retry_state.outcome.result()) 

178 else: 

179 return False 

180 

181 

182class retry_if_not_result(retry_base): 

183 """Retries if the result refutes a predicate.""" 

184 

185 def __init__(self, predicate: typing.Callable[[typing.Any], bool]) -> None: 

186 self.predicate = predicate 

187 

188 def __call__(self, retry_state: "RetryCallState") -> bool: 

189 if retry_state.outcome is None: 

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

191 

192 if not retry_state.outcome.failed: 

193 return not self.predicate(retry_state.outcome.result()) 

194 else: 

195 return False 

196 

197 

198class retry_if_exception_message(retry_if_exception): 

199 """Retries if an exception message equals or matches.""" 

200 

201 def __init__( 

202 self, 

203 message: typing.Optional[str] = None, 

204 match: typing.Optional[str] = None, 

205 ) -> None: 

206 if message and match: 

207 raise TypeError(f"{self.__class__.__name__}() takes either 'message' or 'match', not both") 

208 

209 # set predicate 

210 if message: 

211 

212 def message_fnc(exception: BaseException) -> bool: 

213 return message == str(exception) 

214 

215 predicate = message_fnc 

216 elif match: 

217 prog = re.compile(match) 

218 

219 def match_fnc(exception: BaseException) -> bool: 

220 return bool(prog.match(str(exception))) 

221 

222 predicate = match_fnc 

223 else: 

224 raise TypeError(f"{self.__class__.__name__}() missing 1 required argument 'message' or 'match'") 

225 

226 super().__init__(predicate) 

227 

228 

229class retry_if_not_exception_message(retry_if_exception_message): 

230 """Retries until an exception message equals or matches.""" 

231 

232 def __init__( 

233 self, 

234 message: typing.Optional[str] = None, 

235 match: typing.Optional[str] = None, 

236 ) -> None: 

237 super().__init__(message, match) 

238 # invert predicate 

239 if_predicate = self.predicate 

240 self.predicate = lambda *args_, **kwargs_: not if_predicate(*args_, **kwargs_) 

241 

242 def __call__(self, retry_state: "RetryCallState") -> bool: 

243 if retry_state.outcome is None: 

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

245 

246 if not retry_state.outcome.failed: 

247 return True 

248 

249 exception = retry_state.outcome.exception() 

250 if exception is None: 

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

252 return self.predicate(exception) 

253 

254 

255class retry_any(retry_base): 

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

257 

258 def __init__(self, *retries: retry_base) -> None: 

259 self.retries = retries 

260 

261 def __call__(self, retry_state: "RetryCallState") -> bool: 

262 return any(r(retry_state) for r in self.retries) 

263 

264 

265class retry_all(retry_base): 

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

267 

268 def __init__(self, *retries: retry_base) -> None: 

269 self.retries = retries 

270 

271 def __call__(self, retry_state: "RetryCallState") -> bool: 

272 return all(r(retry_state) for r in self.retries)