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.16import abc17import typing18 19from tenacity import _utils20from tenacity import retry_base21 22if typing.TYPE_CHECKING:23    from tenacity import RetryCallState24 25 26class async_retry_base(retry_base):27    """Abstract base class for async retry strategies."""28 29    @abc.abstractmethod30    async def __call__(self, retry_state: "RetryCallState") -> bool:  # type: ignore[override]31        pass32 33    def __and__(  # type: ignore[override]34        self, other: "typing.Union[retry_base, async_retry_base]"35    ) -> "retry_all":36        return retry_all(self, other)37 38    def __rand__(  # type: ignore[misc,override]39        self, other: "typing.Union[retry_base, async_retry_base]"40    ) -> "retry_all":41        return retry_all(other, self)42 43    def __or__(  # type: ignore[override]44        self, other: "typing.Union[retry_base, async_retry_base]"45    ) -> "retry_any":46        return retry_any(self, other)47 48    def __ror__(  # type: ignore[misc,override]49        self, other: "typing.Union[retry_base, async_retry_base]"50    ) -> "retry_any":51        return retry_any(other, self)52 53 54RetryBaseT = typing.Union[55    async_retry_base, typing.Callable[["RetryCallState"], typing.Awaitable[bool]]56]57 58 59class retry_if_exception(async_retry_base):60    """Retry strategy that retries if an exception verifies a predicate."""61 62    def __init__(63        self, predicate: typing.Callable[[BaseException], typing.Awaitable[bool]]64    ) -> None:65        self.predicate = predicate66 67    async def __call__(self, retry_state: "RetryCallState") -> bool:  # type: ignore[override]68        if retry_state.outcome is None:69            raise RuntimeError("__call__() called before outcome was set")70 71        if retry_state.outcome.failed:72            exception = retry_state.outcome.exception()73            if exception is None:74                raise RuntimeError("outcome failed but the exception is None")75            return await self.predicate(exception)76        else:77            return False78 79 80class retry_if_result(async_retry_base):81    """Retries if the result verifies a predicate."""82 83    def __init__(84        self, predicate: typing.Callable[[typing.Any], typing.Awaitable[bool]]85    ) -> None:86        self.predicate = predicate87 88    async def __call__(self, retry_state: "RetryCallState") -> bool:  # type: ignore[override]89        if retry_state.outcome is None:90            raise RuntimeError("__call__() called before outcome was set")91 92        if not retry_state.outcome.failed:93            return await self.predicate(retry_state.outcome.result())94        else:95            return False96 97 98class retry_any(async_retry_base):99    """Retries if any of the retries condition is valid."""100 101    def __init__(self, *retries: typing.Union[retry_base, async_retry_base]) -> None:102        self.retries = retries103 104    async def __call__(self, retry_state: "RetryCallState") -> bool:  # type: ignore[override]105        result = False106        for r in self.retries:107            result = result or await _utils.wrap_to_async_func(r)(retry_state)108            if result:109                break110        return result111 112 113class retry_all(async_retry_base):114    """Retries if all the retries condition are valid."""115 116    def __init__(self, *retries: typing.Union[retry_base, async_retry_base]) -> None:117        self.retries = retries118 119    async def __call__(self, retry_state: "RetryCallState") -> bool:  # type: ignore[override]120        result = True121        for r in self.retries:122            result = result and await _utils.wrap_to_async_func(r)(retry_state)123            if not result:124                break125        return result126