Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/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 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( 

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

209 ) 

210 

211 # set predicate 

212 if message: 

213 

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

215 return message == str(exception) 

216 

217 predicate = message_fnc 

218 elif match: 

219 prog = re.compile(match) 

220 

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

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

223 

224 predicate = match_fnc 

225 else: 

226 raise TypeError( 

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

228 ) 

229 

230 super().__init__(predicate) 

231 

232 

233class retry_if_not_exception_message(retry_if_exception_message): 

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

235 

236 def __init__( 

237 self, 

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

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

240 ) -> None: 

241 super().__init__(message, match) 

242 # invert predicate 

243 if_predicate = self.predicate 

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

245 

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

247 if retry_state.outcome is None: 

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

249 

250 if not retry_state.outcome.failed: 

251 return True 

252 

253 exception = retry_state.outcome.exception() 

254 if exception is None: 

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

256 return self.predicate(exception) 

257 

258 

259class retry_any(retry_base): 

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

261 

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

263 self.retries = retries 

264 

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

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

267 

268 

269class retry_all(retry_base): 

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

271 

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

273 self.retries = retries 

274 

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

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