codekingpro/portable-devtools

1# Copyright 2016-2018 Julien Danjou2# Copyright 2017 Elisey Zanko3# Copyright 2016 Étienne Bersac4# Copyright 2016 Joshua Harlow5# Copyright 2013-2014 Ray Holder6#7# Licensed under the Apache License, Version 2.0 (the "License");8# you may not use this file except in compliance with the License.9# You may obtain a copy of the License at10#11# http://www.apache.org/licenses/LICENSE-2.012#13# Unless required by applicable law or agreed to in writing, software14# distributed under the License is distributed on an "AS IS" BASIS,15# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.16# See the License for the specific language governing permissions and17# limitations under the License.18import dataclasses19import functools20import sys21import threading22import time23import types24import typing as t25import warnings26from abc import ABC, abstractmethod27from concurrent import futures28 29from . import _utils30 31# Import all built-in retry strategies for easier usage.32from .retry import retry_base  # noqa33from .retry import retry_all  # noqa34from .retry import retry_always  # noqa35from .retry import retry_any  # noqa36from .retry import retry_if_exception  # noqa37from .retry import retry_if_exception_type  # noqa38from .retry import retry_if_exception_cause_type  # noqa39from .retry import retry_if_not_exception_type  # noqa40from .retry import retry_if_not_result  # noqa41from .retry import retry_if_result  # noqa42from .retry import retry_never  # noqa43from .retry import retry_unless_exception_type  # noqa44from .retry import retry_if_exception_message  # noqa45from .retry import retry_if_not_exception_message  # noqa46 47# Import all nap strategies for easier usage.48from .nap import sleep  # noqa49from .nap import sleep_using_event  # noqa50 51# Import all built-in stop strategies for easier usage.52from .stop import stop_after_attempt  # noqa53from .stop import stop_after_delay  # noqa54from .stop import stop_before_delay  # noqa55from .stop import stop_all  # noqa56from .stop import stop_any  # noqa57from .stop import stop_never  # noqa58from .stop import stop_when_event_set  # noqa59 60# Import all built-in wait strategies for easier usage.61from .wait import wait_chain  # noqa62from .wait import wait_combine  # noqa63from .wait import wait_exception  # noqa64from .wait import wait_exponential  # noqa65from .wait import wait_fixed  # noqa66from .wait import wait_incrementing  # noqa67from .wait import wait_none  # noqa68from .wait import wait_random  # noqa69from .wait import wait_random_exponential  # noqa70from .wait import wait_random_exponential as wait_full_jitter  # noqa71from .wait import wait_exponential_jitter  # noqa72 73# Import all built-in before strategies for easier usage.74from .before import before_log  # noqa75from .before import before_nothing  # noqa76 77# Import all built-in after strategies for easier usage.78from .after import after_log  # noqa79from .after import after_nothing  # noqa80 81# Import all built-in before sleep strategies for easier usage.82from .before_sleep import before_sleep_log  # noqa83from .before_sleep import before_sleep_nothing  # noqa84 85try:86    import tornado87except ImportError:88    tornado = None89 90if t.TYPE_CHECKING:91    from typing_extensions import Self92 93    from . import asyncio as tasyncio94    from .retry import RetryBaseT95    from .stop import StopBaseT96    from .wait import WaitBaseT97 98 99WrappedFnReturnT = t.TypeVar("WrappedFnReturnT")100WrappedFn = t.TypeVar("WrappedFn", bound=t.Callable[..., t.Any])101P = t.ParamSpec("P")102R = t.TypeVar("R")103 104 105@dataclasses.dataclass(slots=True)106class IterState:107    actions: t.List[t.Callable[["RetryCallState"], t.Any]] = dataclasses.field(108        default_factory=list109    )110    retry_run_result: bool = False111    delay_since_first_attempt: int = 0112    stop_run_result: bool = False113    is_explicit_retry: bool = False114 115    def reset(self) -> None:116        self.actions = []117        self.retry_run_result = False118        self.delay_since_first_attempt = 0119        self.stop_run_result = False120        self.is_explicit_retry = False121 122 123class TryAgain(Exception):124    """Always retry the executed function when raised."""125 126 127NO_RESULT = object()128 129 130class DoAttempt:131    pass132 133 134class DoSleep(float):135    pass136 137 138class BaseAction:139    """Base class for representing actions to take by retry object.140 141    Concrete implementations must define:142    - __init__: to initialize all necessary fields143    - REPR_FIELDS: class variable specifying attributes to include in repr(self)144    - NAME: for identification in retry object methods and callbacks145    """146 147    REPR_FIELDS: t.Sequence[str] = ()148    NAME: t.Optional[str] = None149 150    def __repr__(self) -> str:151        state_str = ", ".join(152            f"{field}={getattr(self, field)!r}" for field in self.REPR_FIELDS153        )154        return f"{self.__class__.__name__}({state_str})"155 156    def __str__(self) -> str:157        return repr(self)158 159 160class RetryAction(BaseAction):161    REPR_FIELDS = ("sleep",)162    NAME = "retry"163 164    def __init__(self, sleep: t.SupportsFloat) -> None:165        self.sleep = float(sleep)166 167 168_unset = object()169 170 171def _first_set(first: t.Union[t.Any, object], second: t.Any) -> t.Any:172    return second if first is _unset else first173 174 175class RetryError(Exception):176    """Encapsulates the last attempt instance right before giving up."""177 178    def __init__(self, last_attempt: "Future") -> None:179        self.last_attempt = last_attempt180        super().__init__(last_attempt)181 182    def reraise(self) -> t.NoReturn:183        if self.last_attempt.failed:184            raise self.last_attempt.result()185        raise self186 187    def __str__(self) -> str:188        return f"{self.__class__.__name__}[{self.last_attempt}]"189 190 191class AttemptManager:192    """Manage attempt context."""193 194    def __init__(self, retry_state: "RetryCallState"):195        self.retry_state = retry_state196 197    def __enter__(self) -> None:198        pass199 200    def __exit__(201        self,202        exc_type: t.Optional[t.Type[BaseException]],203        exc_value: t.Optional[BaseException],204        traceback: t.Optional["types.TracebackType"],205    ) -> t.Optional[bool]:206        if exc_type is not None and exc_value is not None:207            self.retry_state.set_exception((exc_type, exc_value, traceback))208            return True  # Swallow exception.209        else:210            # We don't have the result, actually.211            self.retry_state.set_result(None)212            return None213 214 215class BaseRetrying(ABC):216    def __init__(217        self,218        sleep: t.Callable[[t.Union[int, float]], None] = sleep,219        stop: "StopBaseT" = stop_never,220        wait: "WaitBaseT" = wait_none(),221        retry: "RetryBaseT" = retry_if_exception_type(),222        before: t.Callable[["RetryCallState"], None] = before_nothing,223        after: t.Callable[["RetryCallState"], None] = after_nothing,224        before_sleep: t.Optional[t.Callable[["RetryCallState"], None]] = None,225        reraise: bool = False,226        retry_error_cls: t.Type[RetryError] = RetryError,227        retry_error_callback: t.Optional[t.Callable[["RetryCallState"], t.Any]] = None,228    ):229        self.sleep = sleep230        self.stop = stop231        self.wait = wait232        self.retry = retry233        self.before = before234        self.after = after235        self.before_sleep = before_sleep236        self.reraise = reraise237        self._local = threading.local()238        self.retry_error_cls = retry_error_cls239        self.retry_error_callback = retry_error_callback240 241    def copy(242        self,243        sleep: t.Union[t.Callable[[t.Union[int, float]], None], object] = _unset,244        stop: t.Union["StopBaseT", object] = _unset,245        wait: t.Union["WaitBaseT", object] = _unset,246        retry: t.Union[retry_base, object] = _unset,247        before: t.Union[t.Callable[["RetryCallState"], None], object] = _unset,248        after: t.Union[t.Callable[["RetryCallState"], None], object] = _unset,249        before_sleep: t.Union[250            t.Optional[t.Callable[["RetryCallState"], None]], object251        ] = _unset,252        reraise: t.Union[bool, object] = _unset,253        retry_error_cls: t.Union[t.Type[RetryError], object] = _unset,254        retry_error_callback: t.Union[255            t.Optional[t.Callable[["RetryCallState"], t.Any]], object256        ] = _unset,257    ) -> "Self":258        """Copy this object with some parameters changed if needed."""259        return self.__class__(260            sleep=_first_set(sleep, self.sleep),261            stop=_first_set(stop, self.stop),262            wait=_first_set(wait, self.wait),263            retry=_first_set(retry, self.retry),264            before=_first_set(before, self.before),265            after=_first_set(after, self.after),266            before_sleep=_first_set(before_sleep, self.before_sleep),267            reraise=_first_set(reraise, self.reraise),268            retry_error_cls=_first_set(retry_error_cls, self.retry_error_cls),269            retry_error_callback=_first_set(270                retry_error_callback, self.retry_error_callback271            ),272        )273 274    def __repr__(self) -> str:275        return (276            f"<{self.__class__.__name__} object at 0x{id(self):x} ("277            f"stop={self.stop}, "278            f"wait={self.wait}, "279            f"sleep={self.sleep}, "280            f"retry={self.retry}, "281            f"before={self.before}, "282            f"after={self.after})>"283        )284 285    @property286    def statistics(self) -> t.Dict[str, t.Any]:287        """Return a dictionary of runtime statistics.288 289        This dictionary will be empty when the controller has never been290        ran. When it is running or has ran previously it should have (but291        may not) have useful and/or informational keys and values when292        running is underway and/or completed.293 294        .. warning:: The keys in this dictionary **should** be some what295                     stable (not changing), but there existence **may**296                     change between major releases as new statistics are297                     gathered or removed so before accessing keys ensure that298                     they actually exist and handle when they do not.299 300        .. note:: The values in this dictionary are local to the thread301                  running call (so if multiple threads share the same retrying302                  object - either directly or indirectly) they will each have303                  there own view of statistics they have collected (in the304                  future we may provide a way to aggregate the various305                  statistics from each thread).306        """307        if not hasattr(self._local, "statistics"):308            self._local.statistics = t.cast(t.Dict[str, t.Any], {})309        return self._local.statistics  # type: ignore[no-any-return]310 311    @property312    def iter_state(self) -> IterState:313        if not hasattr(self._local, "iter_state"):314            self._local.iter_state = IterState()315        return self._local.iter_state  # type: ignore[no-any-return]316 317    def wraps(self, f: WrappedFn) -> WrappedFn:318        """Wrap a function for retrying.319 320        :param f: A function to wraps for retrying.321        """322 323        @functools.wraps(324            f, functools.WRAPPER_ASSIGNMENTS + ("__defaults__", "__kwdefaults__")325        )326        def wrapped_f(*args: t.Any, **kw: t.Any) -> t.Any:327            # Always create a copy to prevent overwriting the local contexts when328            # calling the same wrapped functions multiple times in the same stack329            copy = self.copy()330            wrapped_f.statistics = copy.statistics  # type: ignore[attr-defined]331            return copy(f, *args, **kw)332 333        def retry_with(*args: t.Any, **kwargs: t.Any) -> WrappedFn:334            return self.copy(*args, **kwargs).wraps(f)335 336        # Preserve attributes337        wrapped_f.retry = self  # type: ignore[attr-defined]338        wrapped_f.retry_with = retry_with  # type: ignore[attr-defined]339        wrapped_f.statistics = {}  # type: ignore[attr-defined]340 341        return wrapped_f  # type: ignore[return-value]342 343    def begin(self) -> None:344        self.statistics.clear()345        self.statistics["start_time"] = time.monotonic()346        self.statistics["attempt_number"] = 1347        self.statistics["idle_for"] = 0348 349    def _add_action_func(self, fn: t.Callable[..., t.Any]) -> None:350        self.iter_state.actions.append(fn)351 352    def _run_retry(self, retry_state: "RetryCallState") -> None:353        self.iter_state.retry_run_result = self.retry(retry_state)354 355    def _run_wait(self, retry_state: "RetryCallState") -> None:356        if self.wait:357            sleep = self.wait(retry_state)358        else:359            sleep = 0.0360 361        retry_state.upcoming_sleep = sleep362 363    def _run_stop(self, retry_state: "RetryCallState") -> None:364        self.statistics["delay_since_first_attempt"] = retry_state.seconds_since_start365        self.iter_state.stop_run_result = self.stop(retry_state)366 367    def iter(self, retry_state: "RetryCallState") -> t.Union[DoAttempt, DoSleep, t.Any]:  # noqa368        self._begin_iter(retry_state)369        result = None370        for action in self.iter_state.actions:371            result = action(retry_state)372        return result373 374    def _begin_iter(self, retry_state: "RetryCallState") -> None:  # noqa375        self.iter_state.reset()376 377        fut = retry_state.outcome378        if fut is None:379            if self.before is not None:380                self._add_action_func(self.before)381            self._add_action_func(lambda rs: DoAttempt())382            return383 384        self.iter_state.is_explicit_retry = fut.failed and isinstance(385            fut.exception(), TryAgain386        )387        if not self.iter_state.is_explicit_retry:388            self._add_action_func(self._run_retry)389        self._add_action_func(self._post_retry_check_actions)390 391    def _post_retry_check_actions(self, retry_state: "RetryCallState") -> None:392        if not (self.iter_state.is_explicit_retry or self.iter_state.retry_run_result):393            self._add_action_func(lambda rs: rs.outcome.result())394            return395 396        if self.after is not None:397            self._add_action_func(self.after)398 399        self._add_action_func(self._run_wait)400        self._add_action_func(self._run_stop)401        self._add_action_func(self._post_stop_check_actions)402 403    def _post_stop_check_actions(self, retry_state: "RetryCallState") -> None:404        if self.iter_state.stop_run_result:405            if self.retry_error_callback:406                self._add_action_func(self.retry_error_callback)407                return408 409            def exc_check(rs: "RetryCallState") -> None:410                fut = t.cast(Future, rs.outcome)411                retry_exc = self.retry_error_cls(fut)412                if self.reraise:413                    raise retry_exc.reraise()414                raise retry_exc from fut.exception()415 416            self._add_action_func(exc_check)417            return418 419        def next_action(rs: "RetryCallState") -> None:420            sleep = rs.upcoming_sleep421            rs.next_action = RetryAction(sleep)422            rs.idle_for += sleep423            self.statistics["idle_for"] += sleep424            self.statistics["attempt_number"] += 1425 426        self._add_action_func(next_action)427 428        if self.before_sleep is not None:429            self._add_action_func(self.before_sleep)430 431        self._add_action_func(lambda rs: DoSleep(rs.upcoming_sleep))432 433    def __iter__(self) -> t.Generator[AttemptManager, None, None]:434        self.begin()435 436        retry_state = RetryCallState(self, fn=None, args=(), kwargs={})437        while True:438            do = self.iter(retry_state=retry_state)439            if isinstance(do, DoAttempt):440                yield AttemptManager(retry_state=retry_state)441            elif isinstance(do, DoSleep):442                retry_state.prepare_for_next_attempt()443                self.sleep(do)444            else:445                break446 447    @abstractmethod448    def __call__(449        self,450        fn: t.Callable[..., WrappedFnReturnT],451        *args: t.Any,452        **kwargs: t.Any,453    ) -> WrappedFnReturnT:454        pass455 456 457class Retrying(BaseRetrying):458    """Retrying controller."""459 460    def __call__(461        self,462        fn: t.Callable[..., WrappedFnReturnT],463        *args: t.Any,464        **kwargs: t.Any,465    ) -> WrappedFnReturnT:466        self.begin()467 468        retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs)469        while True:470            do = self.iter(retry_state=retry_state)471            if isinstance(do, DoAttempt):472                try:473                    result = fn(*args, **kwargs)474                except BaseException:  # noqa: B902475                    retry_state.set_exception(sys.exc_info())  # type: ignore[arg-type]476                else:477                    retry_state.set_result(result)478            elif isinstance(do, DoSleep):479                retry_state.prepare_for_next_attempt()480                self.sleep(do)481            else:482                return do  # type: ignore[no-any-return]483 484 485class Future(futures.Future[t.Any]):486    """Encapsulates a (future or past) attempted call to a target function."""487 488    def __init__(self, attempt_number: int) -> None:489        super().__init__()490        self.attempt_number = attempt_number491 492    @property493    def failed(self) -> bool:494        """Return whether a exception is being held in this future."""495        return self.exception() is not None496 497    @classmethod498    def construct(499        cls, attempt_number: int, value: t.Any, has_exception: bool500    ) -> "Future":501        """Construct a new Future object."""502        fut = cls(attempt_number)503        if has_exception:504            fut.set_exception(value)505        else:506            fut.set_result(value)507        return fut508 509 510class RetryCallState:511    """State related to a single call wrapped with Retrying."""512 513    def __init__(514        self,515        retry_object: BaseRetrying,516        fn: t.Optional[WrappedFn],517        args: t.Any,518        kwargs: t.Any,519    ) -> None:520        #: Retry call start timestamp521        self.start_time = time.monotonic()522        #: Retry manager object523        self.retry_object = retry_object524        #: Function wrapped by this retry call525        self.fn = fn526        #: Arguments of the function wrapped by this retry call527        self.args = args528        #: Keyword arguments of the function wrapped by this retry call529        self.kwargs = kwargs530 531        #: The number of the current attempt532        self.attempt_number: int = 1533        #: Last outcome (result or exception) produced by the function534        self.outcome: t.Optional[Future] = None535        #: Timestamp of the last outcome536        self.outcome_timestamp: t.Optional[float] = None537        #: Time spent sleeping in retries538        self.idle_for: float = 0.0539        #: Next action as decided by the retry manager540        self.next_action: t.Optional[RetryAction] = None541        #: Next sleep time as decided by the retry manager.542        self.upcoming_sleep: float = 0.0543 544    @property545    def seconds_since_start(self) -> t.Optional[float]:546        if self.outcome_timestamp is None:547            return None548        return self.outcome_timestamp - self.start_time549 550    def prepare_for_next_attempt(self) -> None:551        self.outcome = None552        self.outcome_timestamp = None553        self.attempt_number += 1554        self.next_action = None555 556    def set_result(self, val: t.Any) -> None:557        ts = time.monotonic()558        fut = Future(self.attempt_number)559        fut.set_result(val)560        self.outcome, self.outcome_timestamp = fut, ts561 562    def set_exception(563        self,564        exc_info: t.Tuple[565            t.Type[BaseException], BaseException, "types.TracebackType| None"566        ],567    ) -> None:568        ts = time.monotonic()569        fut = Future(self.attempt_number)570        fut.set_exception(exc_info[1])571        self.outcome, self.outcome_timestamp = fut, ts572 573    def __repr__(self) -> str:574        if self.outcome is None:575            result = "none yet"576        elif self.outcome.failed:577            exception = self.outcome.exception()578            result = f"failed ({exception.__class__.__name__} {exception})"579        else:580            result = f"returned {self.outcome.result()}"581 582        slept = float(round(self.idle_for, 2))583        clsname = self.__class__.__name__584        return f"<{clsname} {id(self)}: attempt #{self.attempt_number}; slept for {slept}; last result: {result}>"585 586 587class _AsyncRetryDecorator(t.Protocol):588    @t.overload589    def __call__(590        self, fn: "t.Callable[P, types.CoroutineType[t.Any, t.Any, R]]"591    ) -> "t.Callable[P, types.CoroutineType[t.Any, t.Any, R]]": ...592    @t.overload593    def __call__(594        self, fn: t.Callable[P, t.Coroutine[t.Any, t.Any, R]]595    ) -> t.Callable[P, t.Coroutine[t.Any, t.Any, R]]: ...596    @t.overload597    def __call__(598        self, fn: t.Callable[P, t.Awaitable[R]]599    ) -> t.Callable[P, t.Awaitable[R]]: ...600    @t.overload601    def __call__(self, fn: t.Callable[P, R]) -> t.Callable[P, t.Awaitable[R]]: ...602 603 604@t.overload605def retry(func: WrappedFn) -> WrappedFn: ...606 607 608@t.overload609def retry(610    *,611    sleep: t.Callable[[t.Union[int, float]], t.Awaitable[None]],612    stop: "StopBaseT" = ...,613    wait: "WaitBaseT" = ...,614    retry: "t.Union[RetryBaseT, tasyncio.retry.RetryBaseT]" = ...,615    before: t.Callable[["RetryCallState"], t.Union[None, t.Awaitable[None]]] = ...,616    after: t.Callable[["RetryCallState"], t.Union[None, t.Awaitable[None]]] = ...,617    before_sleep: t.Optional[618        t.Callable[["RetryCallState"], t.Union[None, t.Awaitable[None]]]619    ] = ...,620    reraise: bool = ...,621    retry_error_cls: t.Type["RetryError"] = ...,622    retry_error_callback: t.Optional[623        t.Callable[["RetryCallState"], t.Union[t.Any, t.Awaitable[t.Any]]]624    ] = ...,625) -> _AsyncRetryDecorator: ...626 627 628@t.overload629def retry(630    sleep: t.Callable[[t.Union[int, float]], None] = sleep,631    stop: "StopBaseT" = stop_never,632    wait: "WaitBaseT" = wait_none(),633    retry: "t.Union[RetryBaseT, tasyncio.retry.RetryBaseT]" = retry_if_exception_type(),634    before: t.Callable[635        ["RetryCallState"], t.Union[None, t.Awaitable[None]]636    ] = before_nothing,637    after: t.Callable[638        ["RetryCallState"], t.Union[None, t.Awaitable[None]]639    ] = after_nothing,640    before_sleep: t.Optional[641        t.Callable[["RetryCallState"], t.Union[None, t.Awaitable[None]]]642    ] = None,643    reraise: bool = False,644    retry_error_cls: t.Type["RetryError"] = RetryError,645    retry_error_callback: t.Optional[646        t.Callable[["RetryCallState"], t.Union[t.Any, t.Awaitable[t.Any]]]647    ] = None,648) -> t.Callable[[WrappedFn], WrappedFn]: ...649 650 651def retry(*dargs: t.Any, **dkw: t.Any) -> t.Any:652    """Wrap a function with a new `Retrying` object.653 654    :param dargs: positional arguments passed to Retrying object655    :param dkw: keyword arguments passed to the Retrying object656    """657    # support both @retry and @retry() as valid syntax658    if len(dargs) == 1 and callable(dargs[0]):659        return retry()(dargs[0])660    else:661 662        def wrap(f: WrappedFn) -> WrappedFn:663            if isinstance(f, retry_base):664                warnings.warn(665                    f"Got retry_base instance ({f.__class__.__name__}) as callable argument, "666                    f"this will probably hang indefinitely (did you mean retry={f.__class__.__name__}(...)?)"667                )668            r: "BaseRetrying"669            sleep = dkw.get("sleep")670            if _utils.is_coroutine_callable(f) or (671                sleep is not None and _utils.is_coroutine_callable(sleep)672            ):673                r = AsyncRetrying(*dargs, **dkw)674            elif (675                tornado676                and hasattr(tornado.gen, "is_coroutine_function")677                and tornado.gen.is_coroutine_function(f)678            ):679                r = TornadoRetrying(*dargs, **dkw)680            else:681                r = Retrying(*dargs, **dkw)682 683            return r.wraps(f)684 685        return wrap686 687 688from tenacity.asyncio import AsyncRetrying  # noqa:E402,I100689 690if tornado:691    from tenacity.tornadoweb import TornadoRetrying692 693 694__all__ = [695    "retry_base",696    "retry_all",697    "retry_always",698    "retry_any",699    "retry_if_exception",700    "retry_if_exception_type",701    "retry_if_exception_cause_type",702    "retry_if_not_exception_type",703    "retry_if_not_result",704    "retry_if_result",705    "retry_never",706    "retry_unless_exception_type",707    "retry_if_exception_message",708    "retry_if_not_exception_message",709    "sleep",710    "sleep_using_event",711    "stop_after_attempt",712    "stop_after_delay",713    "stop_before_delay",714    "stop_all",715    "stop_any",716    "stop_never",717    "stop_when_event_set",718    "wait_chain",719    "wait_combine",720    "wait_exception",721    "wait_exponential",722    "wait_fixed",723    "wait_incrementing",724    "wait_none",725    "wait_random",726    "wait_random_exponential",727    "wait_full_jitter",728    "wait_exponential_jitter",729    "before_log",730    "before_nothing",731    "after_log",732    "after_nothing",733    "before_sleep_log",734    "before_sleep_nothing",735    "retry",736    "WrappedFn",737    "TryAgain",738    "NO_RESULT",739    "DoAttempt",740    "DoSleep",741    "BaseAction",742    "RetryAction",743    "RetryError",744    "AttemptManager",745    "BaseRetrying",746    "Retrying",747    "Future",748    "RetryCallState",749    "AsyncRetrying",750]751