D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
alt
/
python311
/
lib
/
python3.11
/
site-packages
/
pip
/
_vendor
/
tenacity
/
Filename :
retry.py
back
Copy
# Copyright 2016–2021 Julien Danjou # Copyright 2016 Joshua Harlow # Copyright 2013-2014 Ray Holder # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import abc import re import typing if typing.TYPE_CHECKING: from pip._vendor.tenacity import RetryCallState class retry_base(abc.ABC): """Abstract base class for retry strategies.""" @abc.abstractmethod def __call__(self, retry_state: "RetryCallState") -> bool: pass def __and__(self, other: "retry_base") -> "retry_all": return retry_all(self, other) def __or__(self, other: "retry_base") -> "retry_any": return retry_any(self, other) class _retry_never(retry_base): """Retry strategy that never rejects any result.""" def __call__(self, retry_state: "RetryCallState") -> bool: return False retry_never = _retry_never() class _retry_always(retry_base): """Retry strategy that always rejects any result.""" def __call__(self, retry_state: "RetryCallState") -> bool: return True retry_always = _retry_always() class retry_if_exception(retry_base): """Retry strategy that retries if an exception verifies a predicate.""" def __init__(self, predicate: typing.Callable[[BaseException], bool]) -> None: self.predicate = predicate def __call__(self, retry_state: "RetryCallState") -> bool: if retry_state.outcome.failed: return self.predicate(retry_state.outcome.exception()) else: return False class retry_if_exception_type(retry_if_exception): """Retries if an exception has been raised of one or more types.""" def __init__( self, exception_types: typing.Union[ typing.Type[BaseException], typing.Tuple[typing.Type[BaseException], ...], ] = Exception, ) -> None: self.exception_types = exception_types super().__init__(lambda e: isinstance(e, exception_types)) class retry_if_not_exception_type(retry_if_exception): """Retries except an exception has been raised of one or more types.""" def __init__( self, exception_types: typing.Union[ typing.Type[BaseException], typing.Tuple[typing.Type[BaseException], ...], ] = Exception, ) -> None: self.exception_types = exception_types super().__init__(lambda e: not isinstance(e, exception_types)) class retry_unless_exception_type(retry_if_exception): """Retries until an exception is raised of one or more types.""" def __init__( self, exception_types: typing.Union[ typing.Type[BaseException], typing.Tuple[typing.Type[BaseException], ...], ] = Exception, ) -> None: self.exception_types = exception_types super().__init__(lambda e: not isinstance(e, exception_types)) def __call__(self, retry_state: "RetryCallState") -> bool: # always retry if no exception was raised if not retry_state.outcome.failed: return True return self.predicate(retry_state.outcome.exception()) class retry_if_result(retry_base): """Retries if the result verifies a predicate.""" def __init__(self, predicate: typing.Callable[[typing.Any], bool]) -> None: self.predicate = predicate def __call__(self, retry_state: "RetryCallState") -> bool: if not retry_state.outcome.failed: return self.predicate(retry_state.outcome.result()) else: return False class retry_if_not_result(retry_base): """Retries if the result refutes a predicate.""" def __init__(self, predicate: typing.Callable[[typing.Any], bool]) -> None: self.predicate = predicate def __call__(self, retry_state: "RetryCallState") -> bool: if not retry_state.outcome.failed: return not self.predicate(retry_state.outcome.result()) else: return False class retry_if_exception_message(retry_if_exception): """Retries if an exception message equals or matches.""" def __init__( self, message: typing.Optional[str] = None, match: typing.Optional[str] = None, ) -> None: if message and match: raise TypeError(f"{self.__class__.__name__}() takes either 'message' or 'match', not both") # set predicate if message: def message_fnc(exception: BaseException) -> bool: return message == str(exception) predicate = message_fnc elif match: prog = re.compile(match) def match_fnc(exception: BaseException) -> bool: return bool(prog.match(str(exception))) predicate = match_fnc else: raise TypeError(f"{self.__class__.__name__}() missing 1 required argument 'message' or 'match'") super().__init__(predicate) class retry_if_not_exception_message(retry_if_exception_message): """Retries until an exception message equals or matches.""" def __init__( self, message: typing.Optional[str] = None, match: typing.Optional[str] = None, ) -> None: super().__init__(message, match) # invert predicate if_predicate = self.predicate self.predicate = lambda *args_, **kwargs_: not if_predicate(*args_, **kwargs_) def __call__(self, retry_state: "RetryCallState") -> bool: if not retry_state.outcome.failed: return True return self.predicate(retry_state.outcome.exception()) class retry_any(retry_base): """Retries if any of the retries condition is valid.""" def __init__(self, *retries: retry_base) -> None: self.retries = retries def __call__(self, retry_state: "RetryCallState") -> bool: return any(r(retry_state) for r in self.retries) class retry_all(retry_base): """Retries if all the retries condition are valid.""" def __init__(self, *retries: retry_base) -> None: self.retries = retries def __call__(self, retry_state: "RetryCallState") -> bool: return all(r(retry_state) for r in self.retries)