python (3.12.0)
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
17 import abc
18 import re
19 import typing
20
21 if typing.TYPE_CHECKING:
22 from pip._vendor.tenacity import RetryCallState
23
24
25 class ESC[4;38;5;81mretry_base(ESC[4;38;5;149mabcESC[4;38;5;149m.ESC[4;38;5;149mABC):
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
39 RetryBaseT = typing.Union[retry_base, typing.Callable[["RetryCallState"], bool]]
40
41
42 class ESC[4;38;5;81m_retry_never(ESC[4;38;5;149mretry_base):
43 """Retry strategy that never rejects any result."""
44
45 def __call__(self, retry_state: "RetryCallState") -> bool:
46 return False
47
48
49 retry_never = _retry_never()
50
51
52 class ESC[4;38;5;81m_retry_always(ESC[4;38;5;149mretry_base):
53 """Retry strategy that always rejects any result."""
54
55 def __call__(self, retry_state: "RetryCallState") -> bool:
56 return True
57
58
59 retry_always = _retry_always()
60
61
62 class ESC[4;38;5;81mretry_if_exception(ESC[4;38;5;149mretry_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
81 class ESC[4;38;5;81mretry_if_exception_type(ESC[4;38;5;149mretry_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
95 class ESC[4;38;5;81mretry_if_not_exception_type(ESC[4;38;5;149mretry_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
109 class ESC[4;38;5;81mretry_unless_exception_type(ESC[4;38;5;149mretry_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
136 class ESC[4;38;5;81mretry_if_exception_cause_type(ESC[4;38;5;149mretry_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
166 class ESC[4;38;5;81mretry_if_result(ESC[4;38;5;149mretry_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
182 class ESC[4;38;5;81mretry_if_not_result(ESC[4;38;5;149mretry_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
198 class ESC[4;38;5;81mretry_if_exception_message(ESC[4;38;5;149mretry_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
229 class ESC[4;38;5;81mretry_if_not_exception_message(ESC[4;38;5;149mretry_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
255 class ESC[4;38;5;81mretry_any(ESC[4;38;5;149mretry_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
265 class ESC[4;38;5;81mretry_all(ESC[4;38;5;149mretry_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)
Browse
Usage
Versions