codekingpro/portable-devtools

1# Copyright 2016–2021 Julien Danjou2# Copyright 2016 Joshua Harlow3# Copyright 2013-2014 Ray Holder4#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 at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# 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 and15# limitations under the License.16 17import abc18import re19import typing20 21if typing.TYPE_CHECKING:22    from tenacity import RetryCallState23 24 25class retry_base(abc.ABC):26    """Abstract base class for retry strategies."""27 28    @abc.abstractmethod29    def __call__(self, retry_state: "RetryCallState") -> bool:30        pass31 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 False53 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 True63 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 = predicate73 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 False85 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_types98        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_types112        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_types126        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 raised133        if not retry_state.outcome.failed:134            return True135 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 finding146    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_types157 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 True167                exc = exc.__cause__168 169        return False170 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 = predicate177 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 False186 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 = predicate193 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 False202 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 predicate218        if message:219 220            def message_fnc(exception: BaseException) -> bool:221                return message == str(exception)222 223            predicate = message_fnc224        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_fnc231        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 predicate249        if_predicate = self.predicate250        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 True258 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 = retries270 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 = retries280 281    def __call__(self, retry_state: "RetryCallState") -> bool:282        return all(r(retry_state) for r in self.retries)283