Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/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 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 other.__rand__(self) 

34 

35 def __rand__(self, other: "retry_base") -> "retry_all": 

36 return retry_all(other, self) 

37 

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

39 return other.__ror__(self) 

40 

41 def __ror__(self, other: "retry_base") -> "retry_any": 

42 return retry_any(other, self) 

43 

44 

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

46 

47 

48class _retry_never(retry_base): 

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

50 

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

52 return False 

53 

54 

55retry_never = _retry_never() 

56 

57 

58class _retry_always(retry_base): 

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

60 

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

62 return True 

63 

64 

65retry_always = _retry_always() 

66 

67 

68class retry_if_exception(retry_base): 

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

70 

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

72 self.predicate = predicate 

73 

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

75 if retry_state.outcome is None: 

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

77 

78 if retry_state.outcome.failed: 

79 exception = retry_state.outcome.exception() 

80 if exception is None: 

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

82 return self.predicate(exception) 

83 else: 

84 return False 

85 

86 

87class retry_if_exception_type(retry_if_exception): 

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

89 

90 def __init__( 

91 self, 

92 exception_types: typing.Union[ 

93 typing.Type[BaseException], 

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

95 ] = Exception, 

96 ) -> None: 

97 self.exception_types = exception_types 

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

99 

100 

101class retry_if_not_exception_type(retry_if_exception): 

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

103 

104 def __init__( 

105 self, 

106 exception_types: typing.Union[ 

107 typing.Type[BaseException], 

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

109 ] = Exception, 

110 ) -> None: 

111 self.exception_types = exception_types 

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

113 

114 

115class retry_unless_exception_type(retry_if_exception): 

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

117 

118 def __init__( 

119 self, 

120 exception_types: typing.Union[ 

121 typing.Type[BaseException], 

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

123 ] = Exception, 

124 ) -> None: 

125 self.exception_types = exception_types 

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

127 

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

129 if retry_state.outcome is None: 

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

131 

132 # always retry if no exception was raised 

133 if not retry_state.outcome.failed: 

134 return True 

135 

136 exception = retry_state.outcome.exception() 

137 if exception is None: 

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

139 return self.predicate(exception) 

140 

141 

142class retry_if_exception_cause_type(retry_base): 

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

144 

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

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

147 """ 

148 

149 def __init__( 

150 self, 

151 exception_types: typing.Union[ 

152 typing.Type[BaseException], 

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

154 ] = Exception, 

155 ) -> None: 

156 self.exception_cause_types = exception_types 

157 

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

159 if retry_state.outcome is None: 

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

161 

162 if retry_state.outcome.failed: 

163 exc = retry_state.outcome.exception() 

164 while exc is not None: 

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

166 return True 

167 exc = exc.__cause__ 

168 

169 return False 

170 

171 

172class retry_if_result(retry_base): 

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

174 

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

176 self.predicate = predicate 

177 

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

179 if retry_state.outcome is None: 

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

181 

182 if not retry_state.outcome.failed: 

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

184 else: 

185 return False 

186 

187 

188class retry_if_not_result(retry_base): 

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

190 

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

192 self.predicate = predicate 

193 

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

195 if retry_state.outcome is None: 

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

197 

198 if not retry_state.outcome.failed: 

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

200 else: 

201 return False 

202 

203 

204class retry_if_exception_message(retry_if_exception): 

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

206 

207 def __init__( 

208 self, 

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

210 match: typing.Union[None, str, typing.Pattern[str]] = None, 

211 ) -> None: 

212 if message and match: 

213 raise TypeError( 

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

215 ) 

216 

217 # set predicate 

218 if message: 

219 

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

221 return message == str(exception) 

222 

223 predicate = message_fnc 

224 elif match: 

225 prog = re.compile(match) 

226 

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

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

229 

230 predicate = match_fnc 

231 else: 

232 raise TypeError( 

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

234 ) 

235 

236 super().__init__(predicate) 

237 

238 

239class retry_if_not_exception_message(retry_if_exception_message): 

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

241 

242 def __init__( 

243 self, 

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

245 match: typing.Union[None, str, typing.Pattern[str]] = None, 

246 ) -> None: 

247 super().__init__(message, match) 

248 # invert predicate 

249 if_predicate = self.predicate 

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

251 

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

253 if retry_state.outcome is None: 

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

255 

256 if not retry_state.outcome.failed: 

257 return True 

258 

259 exception = retry_state.outcome.exception() 

260 if exception is None: 

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

262 return self.predicate(exception) 

263 

264 

265class retry_any(retry_base): 

266 """Retries if any of the retries condition is 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 any(r(retry_state) for r in self.retries) 

273 

274 

275class retry_all(retry_base): 

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

277 

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

279 self.retries = retries 

280 

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

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