Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:53
📄 __init__.py # Copyright 2016-2018 Julien Danjou # Copyright 2017 Elisey Zanko # Copyright 2016 Étienne Bersac # 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 functools import sys import threading import time import typing as t import warnings from abc import ABC, abstractmethod from concurrent import futures from inspect import iscoroutinefunction # Import all built-in retry strategies for easier usage. from .retry import retry_base # noqa from .retry import retry_all # noqa from .retry import retry_always # noqa from .retry import retry_any # noqa from .retry import retry_if_exception # noqa from .retry import retry_if_exception_type # noqa from .retry import retry_if_not_exception_type # noqa from .retry import retry_if_not_result # noqa from .retry import retry_if_result # noqa from .retry import retry_never # noqa from .retry import retry_unless_exception_type # noqa from .retry import retry_if_exception_message # noqa from .retry import retry_if_not_exception_message # noqa # Import all nap strategies for easier usage. from .nap import sleep # noqa from .nap import sleep_using_event # noqa # Import all built-in stop strategies for easier usage. from .stop import stop_after_attempt # noqa from .stop import stop_after_delay # noqa from .stop import stop_all # noqa from .stop import stop_any # noqa from .stop import stop_never # noqa from .stop import stop_when_event_set # noqa # Import all built-in wait strategies for easier usage. from .wait import wait_chain # noqa from .wait import wait_combine # noqa from .wait import wait_exponential # noqa from .wait import wait_fixed # noqa from .wait import wait_incrementing # noqa from .wait import wait_none # noqa from .wait import wait_random # noqa from .wait import wait_random_exponential # noqa from .wait import wait_random_exponential as wait_full_jitter # noqa # Import all built-in before strategies for easier usage. from .before import before_log # noqa from .before import before_nothing # noqa # Import all built-in after strategies for easier usage. from .after import after_log # noqa from .after import after_nothing # noqa # Import all built-in after strategies for easier usage. from .before_sleep import before_sleep_log # noqa from .before_sleep import before_sleep_nothing # noqa # Replace a conditional import with a hard-coded None so that pip does # not attempt to use tornado even if it is present in the environment. # If tornado is non-None, tenacity will attempt to execute some code # that is sensitive to the version of tornado, which could break pip # if an old version is found. tornado = None # type: ignore if t.TYPE_CHECKING: import types from .wait import wait_base from .stop import stop_base WrappedFn = t.TypeVar("WrappedFn", bound=t.Callable) _RetValT = t.TypeVar("_RetValT") @t.overload def retry(fn: WrappedFn) -> WrappedFn: pass @t.overload def retry(dargs: t.Any, dkw: t.Any) -> t.Callable[[WrappedFn], WrappedFn]: # noqa pass def retry(dargs: t.Any, *dkw: t.Any) -> t.Union[WrappedFn, t.Callable[[WrappedFn], WrappedFn]]: # noqa """Wrap a function with a new `Retrying` object. :param dargs: positional arguments passed to Retrying object :param dkw: keyword arguments passed to the Retrying object """ # support both @retry and @retry() as valid syntax if len(dargs) == 1 and callable(dargs[0]): return retry()(dargs[0]) else: def wrap(f: WrappedFn) -> WrappedFn: if isinstance(f, retry_base): warnings.warn( f"Got retry_base instance ({f.__class__.__name__}) as callable argument, " f"this will probably hang indefinitely (did you mean retry={f.__class__.__name__}(...)?)" ) if iscoroutinefunction(f): r: "BaseRetrying" = AsyncRetrying(dargs, *dkw) elif tornado and hasattr(tornado.gen, "is_coroutine_function") and tornado.gen.is_coroutine_function(f): r = TornadoRetrying(dargs, *dkw) else: r = Retrying(dargs, dkw) return r.wraps(f) return wrap class TryAgain(Exception): """Always retry the executed function when raised.""" NO_RESULT = object() class DoAttempt: pass class DoSleep(float): pass class BaseAction: """Base class for representing actions to take by retry object. Concrete implementations must define: - __init__: to initialize all necessary fields - REPR_FIELDS: class variable specifying attributes to include in repr(self) - NAME: for identification in retry object methods and callbacks """ REPR_FIELDS: t.Sequence[str] = () NAME: t.Optional[str] = None def __repr__(self) -> str: state_str = ", ".join(f"{field}={getattr(self, field)!r}" for field in self.REPR_FIELDS) return f"{self.__class__.__name__}({state_str})" def __str__(self) -> str: return repr(self) class RetryAction(BaseAction): REPR_FIELDS = ("sleep",) NAME = "retry" def __init__(self, sleep: t.SupportsFloat) -> None: self.sleep = float(sleep) _unset = object() def _first_set(first: t.Union[t.Any, object], second: t.Any) -> t.Any: return second if first is _unset else first class RetryError(Exception): """Encapsulates the last attempt instance right before giving up.""" def __init__(self, last_attempt: "Future") -> None: self.last_attempt = last_attempt super().__init__(last_attempt) def reraise(self) -> "t.NoReturn": if self.last_attempt.failed: raise self.last_attempt.result() raise self def __str__(self) -> str: return f"{self.__class__.__name__}[{self.last_attempt}]" class AttemptManager: """Manage attempt context.""" def __init__(self, retry_state: "RetryCallState"): self.retry_state = retry_state def __enter__(self) -> None: pass def __exit__( self, exc_type: t.Optional[t.Type[BaseException]], exc_value: t.Optional[BaseException], traceback: t.Optional["types.TracebackType"], ) -> t.Optional[bool]: if isinstance(exc_value, BaseException): self.retry_state.set_exception((exc_type, exc_value, traceback)) return True # Swallow exception. else: # We don't have the result, actually. self.retry_state.set_result(None) return None class BaseRetrying(ABC): def __init__( self, sleep: t.Callable[[t.Union[int, float]], None] = sleep, stop: "stop_base" = stop_never, wait: "wait_base" = wait_none(), retry: retry_base = retry_if_exception_type(), before: t.Callable[["RetryCallState"], None] = before_nothing, after: t.Callable[["RetryCallState"], None] = after_nothing, before_sleep: t.Optional[t.Callable[["RetryCallState"], None]] = None, reraise: bool = False, retry_error_cls: t.Type[RetryError] = RetryError, retry_error_callback: t.Optional[t.Callable[["RetryCallState"], t.Any]] = None, ): self.sleep = sleep self.stop = stop self.wait = wait self.retry = retry self.before = before self.after = after self.before_sleep = before_sleep self.reraise = reraise self._local = threading.local() self.retry_error_cls = retry_error_cls self.retry_error_callback = retry_error_callback def copy( self, sleep: t.Union[t.Callable[[t.Union[int, float]], None], object] = _unset, stop: t.Union["stop_base", object] = _unset, wait: t.Union["wait_base", object] = _unset, retry: t.Union[retry_base, object] = _unset, before: t.Union[t.Callable[["RetryCallState"], None], object] = _unset, after: t.Union[t.Callable[["RetryCallState"], None], object] = _unset, before_sleep: t.Union[t.Optional[t.Callable[["RetryCallState"], None]], object] = _unset, reraise: t.Union[bool, object] = _unset, retry_error_cls: t.Union[t.Type[RetryError], object] = _unset, retry_error_callback: t.Union[t.Optional[t.Callable[["RetryCallState"], t.Any]], object] = _unset, ) -> "BaseRetrying": """Copy this object with some parameters changed if needed.""" return self.__class__( sleep=_first_set(sleep, self.sleep), stop=_first_set(stop, self.stop), wait=_first_set(wait, self.wait), retry=_first_set(retry, self.retry), before=_first_set(before, self.before), after=_first_set(after, self.after), before_sleep=_first_set(before_sleep, self.before_sleep), reraise=_first_set(reraise, self.reraise), retry_error_cls=_first_set(retry_error_cls, self.retry_error_cls), retry_error_callback=_first_set(retry_error_callback, self.retry_error_callback), ) def __repr__(self) -> str: return ( f"<{self.__class__.__name__} object at 0x{id(self):x} (" f"stop={self.stop}, " f"wait={self.wait}, " f"sleep={self.sleep}, " f"retry={self.retry}, " f"before={self.before}, " f"after={self.after})>" ) @property def statistics(self) -> t.Dict[str, t.Any]: """Return a dictionary of runtime statistics. This dictionary will be empty when the controller has never been ran. When it is running or has ran previously it should have (but may not) have useful and/or informational keys and values when running is underway and/or completed. .. warning:: The keys in this dictionary should be some what stable (not changing), but there existence may** change between major releases as new statistics are gathered or removed so before accessing keys ensure that they actually exist and handle when they do not. .. note:: The values in this dictionary are local to the thread running call (so if multiple threads share the same retrying object - either directly or indirectly) they will each have there own view of statistics they have collected (in the future we may provide a way to aggregate the various statistics from each thread). """ try: return self._local.statistics except AttributeError: self._local.statistics = {} return self._local.statistics def wraps(self, f: WrappedFn) -> WrappedFn: """Wrap a function for retrying. :param f: A function to wraps for retrying. """ @functools.wraps(f) def wrapped_f(args: t.Any, kw: t.Any) -> t.Any: return self(f, args, *kw) def retry_with(args: t.Any, *kwargs: t.Any) -> WrappedFn: return self.copy(args, *kwargs).wraps(f) wrapped_f.retry = self wrapped_f.retry_with = retry_with return wrapped_f def begin(self) -> None: self.statistics.clear() self.statistics["start_time"] = time.monotonic() self.statistics["attempt_number"] = 1 self.statistics["idle_for"] = 0 def iter(self, retry_state: "RetryCallState") -> t.Union[DoAttempt, DoSleep, t.Any]: # noqa fut = retry_state.outcome if fut is None: if self.before is not None: self.before(retry_state) return DoAttempt() is_explicit_retry = retry_state.outcome.failed and isinstance(retry_state.outcome.exception(), TryAgain) if not (is_explicit_retry or self.retry(retry_state=retry_state)): return fut.result() if self.after is not None: self.after(retry_state) self.statistics["delay_since_first_attempt"] = retry_state.seconds_since_start if self.stop(retry_state=retry_state): if self.retry_error_callback: return self.retry_error_callback(retry_state) retry_exc = self.retry_error_cls(fut) if self.reraise: raise retry_exc.reraise() raise retry_exc from fut.exception() if self.wait: sleep = self.wait(retry_state=retry_state) else: sleep = 0.0 retry_state.next_action = RetryAction(sleep) retry_state.idle_for += sleep self.statistics["idle_for"] += sleep self.statistics["attempt_number"] += 1 if self.before_sleep is not None: self.before_sleep(retry_state) return DoSleep(sleep) def __iter__(self) -> t.Generator[AttemptManager, None, None]: self.begin() retry_state = RetryCallState(self, fn=None, args=(), kwargs={}) while True: do = self.iter(retry_state=retry_state) if isinstance(do, DoAttempt): yield AttemptManager(retry_state=retry_state) elif isinstance(do, DoSleep): retry_state.prepare_for_next_attempt() self.sleep(do) else: break @abstractmethod def __call__(self, fn: t.Callable[..., _RetValT], args: t.Any, *kwargs: t.Any) -> _RetValT: pass class Retrying(BaseRetrying): """Retrying controller.""" def __call__(self, fn: t.Callable[..., _RetValT], args: t.Any, *kwargs: t.Any) -> _RetValT: self.begin() retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs) while True: do = self.iter(retry_state=retry_state) if isinstance(do, DoAttempt): try: result = fn(args, **kwargs) except BaseException: # noqa: B902 retry_state.set_exception(sys.exc_info()) else: retry_state.set_result(result) elif isinstance(do, DoSleep): retry_state.prepare_for_next_attempt() self.sleep(do) else: return do class Future(futures.Future): """Encapsulates a (future or past) attempted call to a target function.""" def __init__(self, attempt_number: int) -> None: super().__init__() self.attempt_number = attempt_number @property def failed(self) -> bool: """Return whether a exception is being held in this future.""" return self.exception() is not None @classmethod def construct(cls, attempt_number: int, value: t.Any, has_exception: bool) -> "Future": """Construct a new Future object.""" fut = cls(attempt_number) if has_exception: fut.set_exception(value) else: fut.set_result(value) return fut class RetryCallState: """State related to a single call wrapped with Retrying.""" def __init__( self, retry_object: BaseRetrying, fn: t.Optional[WrappedFn], args: t.Any, kwargs: t.Any, ) -> None: #: Retry call start timestamp self.start_time = time.monotonic() #: Retry manager object self.retry_object = retry_object #: Function wrapped by this retry call self.fn = fn #: Arguments of the function wrapped by this retry call self.args = args #: Keyword arguments of the function wrapped by this retry call self.kwargs = kwargs #: The number of the current attempt self.attempt_number: int = 1 #: Last outcome (result or exception) produced by the function self.outcome: t.Optional[Future] = None #: Timestamp of the last outcome self.outcome_timestamp: t.Optional[float] = None #: Time spent sleeping in retries self.idle_for: float = 0.0 #: Next action as decided by the retry manager self.next_action: t.Optional[RetryAction] = None @property def seconds_since_start(self) -> t.Optional[float]: if self.outcome_timestamp is None: return None return self.outcome_timestamp - self.start_time def prepare_for_next_attempt(self) -> None: self.outcome = None self.outcome_timestamp = None self.attempt_number += 1 self.next_action = None def set_result(self, val: t.Any) -> None: ts = time.monotonic() fut = Future(self.attempt_number) fut.set_result(val) self.outcome, self.outcome_timestamp = fut, ts def set_exception(self, exc_info: t.Tuple[t.Type[BaseException], BaseException, "types.TracebackType"]) -> None: ts = time.monotonic() fut = Future(self.attempt_number) fut.set_exception(exc_info[1]) self.outcome, self.outcome_timestamp = fut, ts def __repr__(self): if self.outcome is None: result = "none yet" elif self.outcome.failed: exception = self.outcome.exception() result = f"failed ({exception.__class__.__name__} {exception})" else: result = f"returned {self.outcome.result()}" slept = float(round(self.idle_for, 2)) clsname = self.__class__.__name__ return f"<{clsname} {id(self)}: attempt #{self.attempt_number}; slept for {slept}; last result: {result}>" from pip._vendor.tenacity._asyncio import AsyncRetrying # noqa:E402,I100 if tornado: from pip._vendor.tenacity.tornadoweb import TornadoRetrying	17.83 KB	2026-01-07 23:39
📁 __pycache__	-	2026-03-21 15:53
📄 _asyncio.py # Copyright 2016 Étienne Bersac # Copyright 2016 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 functools import sys import typing from asyncio import sleep from pip._vendor.tenacity import AttemptManager from pip._vendor.tenacity import BaseRetrying from pip._vendor.tenacity import DoAttempt from pip._vendor.tenacity import DoSleep from pip._vendor.tenacity import RetryCallState WrappedFn = typing.TypeVar("WrappedFn", bound=typing.Callable) _RetValT = typing.TypeVar("_RetValT") class AsyncRetrying(BaseRetrying): def __init__(self, sleep: typing.Callable[[float], typing.Awaitable] = sleep, kwargs: typing.Any) -> None: super().__init__(kwargs) self.sleep = sleep async def __call__( # type: ignore # Change signature from supertype self, fn: typing.Callable[..., typing.Awaitable[_RetValT]], args: typing.Any, kwargs: typing.Any, ) -> _RetValT: self.begin() retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs) while True: do = self.iter(retry_state=retry_state) if isinstance(do, DoAttempt): try: result = await fn(args, *kwargs) except BaseException: # noqa: B902 retry_state.set_exception(sys.exc_info()) else: retry_state.set_result(result) elif isinstance(do, DoSleep): retry_state.prepare_for_next_attempt() await self.sleep(do) else: return do def __aiter__(self) -> "AsyncRetrying": self.begin() self._retry_state = RetryCallState(self, fn=None, args=(), kwargs={}) return self async def __anext__(self) -> typing.Union[AttemptManager, typing.Any]: while True: do = self.iter(retry_state=self._retry_state) if do is None: raise StopAsyncIteration elif isinstance(do, DoAttempt): return AttemptManager(retry_state=self._retry_state) elif isinstance(do, DoSleep): self._retry_state.prepare_for_next_attempt() await self.sleep(do) else: return do def wraps(self, fn: WrappedFn) -> WrappedFn: fn = super().wraps(fn) # Ensure wrapper is recognized as a coroutine function. @functools.wraps(fn) async def async_wrapped(args: typing.Any, *kwargs: typing.Any) -> typing.Any: return await fn(args, **kwargs) # Preserve attributes async_wrapped.retry = fn.retry async_wrapped.retry_with = fn.retry_with return async_wrapped	3.24 KB	2026-01-07 23:39
📄 _utils.py # Copyright 2016 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 sys import typing # sys.maxsize: # An integer giving the maximum value a variable of type Py_ssize_t can take. MAX_WAIT = sys.maxsize / 2 def find_ordinal(pos_num: int) -> str: # See: https://en.wikipedia.org/wiki/English_numerals#Ordinal_numbers if pos_num == 0: return "th" elif pos_num == 1: return "st" elif pos_num == 2: return "nd" elif pos_num == 3: return "rd" elif 4 <= pos_num <= 20: return "th" else: return find_ordinal(pos_num % 10) def to_ordinal(pos_num: int) -> str: return f"{pos_num}{find_ordinal(pos_num)}" def get_callback_name(cb: typing.Callable[..., typing.Any]) -> str: """Get a callback fully-qualified name. If no name can be produced ``repr(cb)`` is called and returned. """ segments = [] try: segments.append(cb.__qualname__) except AttributeError: try: segments.append(cb.__name__) except AttributeError: pass if not segments: return repr(cb) else: try: # When running under sphinx it appears this can be none? if cb.__module__: segments.insert(0, cb.__module__) except AttributeError: pass return ".".join(segments)	1.9 KB	2026-01-07 23:39
📄 after.py # Copyright 2016 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 typing from pip._vendor.tenacity import _utils if typing.TYPE_CHECKING: import logging from pip._vendor.tenacity import RetryCallState def after_nothing(retry_state: "RetryCallState") -> None: """After call strategy that does nothing.""" def after_log( logger: "logging.Logger", log_level: int, sec_format: str = "%0.3f", ) -> typing.Callable[["RetryCallState"], None]: """After call strategy that logs to some logger the finished attempt.""" def log_it(retry_state: "RetryCallState") -> None: logger.log( log_level, f"Finished call to '{_utils.get_callback_name(retry_state.fn)}' " f"after {sec_format % retry_state.seconds_since_start}(s), " f"this was the {_utils.to_ordinal(retry_state.attempt_number)} time calling it.", ) return log_it	1.46 KB	2026-01-07 23:39
📄 before.py # Copyright 2016 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 typing from pip._vendor.tenacity import _utils if typing.TYPE_CHECKING: import logging from pip._vendor.tenacity import RetryCallState def before_nothing(retry_state: "RetryCallState") -> None: """Before call strategy that does nothing.""" def before_log(logger: "logging.Logger", log_level: int) -> typing.Callable[["RetryCallState"], None]: """Before call strategy that logs to some logger the attempt.""" def log_it(retry_state: "RetryCallState") -> None: logger.log( log_level, f"Starting call to '{_utils.get_callback_name(retry_state.fn)}', " f"this is the {_utils.to_ordinal(retry_state.attempt_number)} time calling it.", ) return log_it	1.34 KB	2026-01-07 23:39
📄 before_sleep.py # Copyright 2016 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 typing from pip._vendor.tenacity import _utils if typing.TYPE_CHECKING: import logging from pip._vendor.tenacity import RetryCallState def before_sleep_nothing(retry_state: "RetryCallState") -> None: """Before call strategy that does nothing.""" def before_sleep_log( logger: "logging.Logger", log_level: int, exc_info: bool = False, ) -> typing.Callable[["RetryCallState"], None]: """Before call strategy that logs to some logger the attempt.""" def log_it(retry_state: "RetryCallState") -> None: if retry_state.outcome.failed: ex = retry_state.outcome.exception() verb, value = "raised", f"{ex.__class__.__name__}: {ex}" if exc_info: local_exc_info = retry_state.outcome.exception() else: local_exc_info = False else: verb, value = "returned", retry_state.outcome.result() local_exc_info = False # exc_info does not apply when no exception logger.log( log_level, f"Retrying {_utils.get_callback_name(retry_state.fn)} " f"in {retry_state.next_action.sleep} seconds as it {verb} {value}.", exc_info=local_exc_info, ) return log_it	1.86 KB	2026-01-07 23:39
📄 nap.py # Copyright 2016 Étienne Bersac # Copyright 2016 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 time import typing if typing.TYPE_CHECKING: import threading def sleep(seconds: float) -> None: """ Sleep strategy that delays execution for a given number of seconds. This is the default strategy, and may be mocked out for unit testing. """ time.sleep(seconds) class sleep_using_event: """Sleep strategy that waits on an event to be set.""" def __init__(self, event: "threading.Event") -> None: self.event = event def __call__(self, timeout: typing.Optional[float]) -> None: # NOTE(harlowja): this may not actually wait for timeout # seconds if the event is set (ie this may eject out early). self.event.wait(timeout=timeout)	1.35 KB	2026-01-07 23:39
📄 retry.py # 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)	6.49 KB	2026-01-07 23:39
📄 stop.py # 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 typing if typing.TYPE_CHECKING: import threading from pip._vendor.tenacity import RetryCallState class stop_base(abc.ABC): """Abstract base class for stop strategies.""" @abc.abstractmethod def __call__(self, retry_state: "RetryCallState") -> bool: pass def __and__(self, other: "stop_base") -> "stop_all": return stop_all(self, other) def __or__(self, other: "stop_base") -> "stop_any": return stop_any(self, other) class stop_any(stop_base): """Stop if any of the stop condition is valid.""" def __init__(self, stops: stop_base) -> None: self.stops = stops def __call__(self, retry_state: "RetryCallState") -> bool: return any(x(retry_state) for x in self.stops) class stop_all(stop_base): """Stop if all the stop conditions are valid.""" def __init__(self, stops: stop_base) -> None: self.stops = stops def __call__(self, retry_state: "RetryCallState") -> bool: return all(x(retry_state) for x in self.stops) class _stop_never(stop_base): """Never stop.""" def __call__(self, retry_state: "RetryCallState") -> bool: return False stop_never = _stop_never() class stop_when_event_set(stop_base): """Stop when the given event is set.""" def __init__(self, event: "threading.Event") -> None: self.event = event def __call__(self, retry_state: "RetryCallState") -> bool: return self.event.is_set() class stop_after_attempt(stop_base): """Stop when the previous attempt >= max_attempt.""" def __init__(self, max_attempt_number: int) -> None: self.max_attempt_number = max_attempt_number def __call__(self, retry_state: "RetryCallState") -> bool: return retry_state.attempt_number >= self.max_attempt_number class stop_after_delay(stop_base): """Stop when the time from the first attempt >= limit.""" def __init__(self, max_delay: float) -> None: self.max_delay = max_delay def __call__(self, retry_state: "RetryCallState") -> bool: return retry_state.seconds_since_start >= self.max_delay	2.72 KB	2026-01-07 23:39
📄 tornadoweb.py # Copyright 2017 Elisey Zanko # # 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 sys import typing from pip._vendor.tenacity import BaseRetrying from pip._vendor.tenacity import DoAttempt from pip._vendor.tenacity import DoSleep from pip._vendor.tenacity import RetryCallState from tornado import gen if typing.TYPE_CHECKING: from tornado.concurrent import Future _RetValT = typing.TypeVar("_RetValT") class TornadoRetrying(BaseRetrying): def __init__(self, sleep: "typing.Callable[[float], Future[None]]" = gen.sleep, kwargs: typing.Any) -> None: super().__init__(kwargs) self.sleep = sleep @gen.coroutine def __call__( # type: ignore # Change signature from supertype self, fn: "typing.Callable[..., typing.Union[typing.Generator[typing.Any, typing.Any, _RetValT], Future[_RetValT]]]", args: typing.Any, kwargs: typing.Any, ) -> "typing.Generator[typing.Any, typing.Any, _RetValT]": self.begin() retry_state = RetryCallState(retry_object=self, fn=fn, args=args, kwargs=kwargs) while True: do = self.iter(retry_state=retry_state) if isinstance(do, DoAttempt): try: result = yield fn(args, **kwargs) except BaseException: # noqa: B902 retry_state.set_exception(sys.exc_info()) else: retry_state.set_result(result) elif isinstance(do, DoSleep): retry_state.prepare_for_next_attempt() yield self.sleep(do) else: raise gen.Return(do)	2.09 KB	2026-01-07 23:39
📄 wait.py # 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 random import typing from pip._vendor.tenacity import _utils if typing.TYPE_CHECKING: from pip._vendor.tenacity import RetryCallState class wait_base(abc.ABC): """Abstract base class for wait strategies.""" @abc.abstractmethod def __call__(self, retry_state: "RetryCallState") -> float: pass def __add__(self, other: "wait_base") -> "wait_combine": return wait_combine(self, other) def __radd__(self, other: "wait_base") -> typing.Union["wait_combine", "wait_base"]: # make it possible to use multiple waits with the built-in sum function if other == 0: return self return self.__add__(other) class wait_fixed(wait_base): """Wait strategy that waits a fixed amount of time between each retry.""" def __init__(self, wait: float) -> None: self.wait_fixed = wait def __call__(self, retry_state: "RetryCallState") -> float: return self.wait_fixed class wait_none(wait_fixed): """Wait strategy that doesn't wait at all before retrying.""" def __init__(self) -> None: super().__init__(0) class wait_random(wait_base): """Wait strategy that waits a random amount of time between min/max.""" def __init__(self, min: typing.Union[int, float] = 0, max: typing.Union[int, float] = 1) -> None: # noqa self.wait_random_min = min self.wait_random_max = max def __call__(self, retry_state: "RetryCallState") -> float: return self.wait_random_min + (random.random() * (self.wait_random_max - self.wait_random_min)) class wait_combine(wait_base): """Combine several waiting strategies.""" def __init__(self, strategies: wait_base) -> None: self.wait_funcs = strategies def __call__(self, retry_state: "RetryCallState") -> float: return sum(x(retry_state=retry_state) for x in self.wait_funcs) class wait_chain(wait_base): """Chain two or more waiting strategies. If all strategies are exhausted, the very last strategy is used thereafter. For example:: @retry(wait=wait_chain([wait_fixed(1) for i in range(3)] + [wait_fixed(2) for j in range(5)] + [wait_fixed(5) for k in range(4))) def wait_chained(): print("Wait 1s for 3 attempts, 2s for 5 attempts and 5s thereafter.") """ def __init__(self, strategies: wait_base) -> None: self.strategies = strategies def __call__(self, retry_state: "RetryCallState") -> float: wait_func_no = min(max(retry_state.attempt_number, 1), len(self.strategies)) wait_func = self.strategies[wait_func_no - 1] return wait_func(retry_state=retry_state) class wait_incrementing(wait_base): """Wait an incremental amount of time after each attempt. Starting at a starting value and incrementing by a value for each attempt (and restricting the upper limit to some maximum value). """ def __init__( self, start: typing.Union[int, float] = 0, increment: typing.Union[int, float] = 100, max: typing.Union[int, float] = _utils.MAX_WAIT, # noqa ) -> None: self.start = start self.increment = increment self.max = max def __call__(self, retry_state: "RetryCallState") -> float: result = self.start + (self.increment (retry_state.attempt_number - 1)) return max(0, min(result, self.max)) class wait_exponential(wait_base): """Wait strategy that applies exponential backoff. It allows for a customized multiplier and an ability to restrict the upper and lower limits to some maximum and minimum value. The intervals are fixed (i.e. there is no jitter), so this strategy is suitable for balancing retries against latency when a required resource is unavailable for an unknown duration, but not suitable for resolving contention between multiple processes for a shared resource. Use wait_random_exponential for the latter case. """ def __init__( self, multiplier: typing.Union[int, float] = 1, max: typing.Union[int, float] = _utils.MAX_WAIT, # noqa exp_base: typing.Union[int, float] = 2, min: typing.Union[int, float] = 0, # noqa ) -> None: self.multiplier = multiplier self.min = min self.max = max self.exp_base = exp_base def __call__(self, retry_state: "RetryCallState") -> float: try: exp = self.exp_base ** (retry_state.attempt_number - 1) result = self.multiplier * exp except OverflowError: return self.max return max(max(0, self.min), min(result, self.max)) class wait_random_exponential(wait_exponential): """Random wait with exponentially widening window. An exponential backoff strategy used to mediate contention between multiple uncoordinated processes for a shared resource in distributed systems. This is the sense in which "exponential backoff" is meant in e.g. Ethernet networking, and corresponds to the "Full Jitter" algorithm described in this blog post: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/ Each retry occurs at a random time in a geometrically expanding interval. It allows for a custom multiplier and an ability to restrict the upper limit of the random interval to some maximum value. Example:: wait_random_exponential(multiplier=0.5, # initial window 0.5s max=60) # max 60s timeout When waiting for an unavailable resource to become available again, as opposed to trying to resolve contention for a shared resource, the wait_exponential strategy (which uses a fixed interval) may be preferable. """ def __call__(self, retry_state: "RetryCallState") -> float: high = super().__call__(retry_state=retry_state) return random.uniform(0, high)	6.53 KB	2026-01-07 23:39

Terminal