Inheritance diagram for BaseRetrying:

[legend]

Collaboration diagram for BaseRetrying:

Public Member Functions
	__init__ (self, t.Callable[[t.Union[int, float]], None] sleep=sleep, "StopBaseT" stop=stop_never, "WaitBaseT" wait=wait_none(), "RetryBaseT" retry=retry_if_exception_type(), t.Callable[["RetryCallState"], None] before=before_nothing, t.Callable[["RetryCallState"], None] after=after_nothing, t.Optional[t.Callable[["RetryCallState"], None]] before_sleep=None, bool reraise=False, t.Type[RetryError] retry_error_cls=RetryError, t.Optional[t.Callable[["RetryCallState"], t.Any]] retry_error_callback=None)

"BaseRetrying"	copy (self, t.Union[t.Callable[[t.Union[int, float]], None], object] sleep=_unset, t.Union["StopBaseT", object] stop=_unset, t.Union["WaitBaseT", object] wait=_unset, t.Union[retry_base, object] retry=_unset, t.Union[t.Callable[["RetryCallState"], None], object] before=_unset, t.Union[t.Callable[["RetryCallState"], None], object] after=_unset, t.Union[t.Optional[t.Callable[["RetryCallState"], None]], object] before_sleep=_unset, t.Union[bool, object] reraise=_unset, t.Union[t.Type[RetryError], object] retry_error_cls=_unset, t.Union[t.Optional[t.Callable[["RetryCallState"], t.Any]], object] retry_error_callback=_unset)

str	__repr__ (self)

t.Dict[str, t.Any]	statistics (self)

WrappedFn	wraps (self, WrappedFn f)

None	begin (self)

t.Union[DoAttempt, DoSleep, t.Any]	iter (self, "RetryCallState" retry_state)

t.Generator[AttemptManager, None, None]	__iter__ (self)

WrappedFnReturnT	__call__ (self, t.Callable[..., WrappedFnReturnT] fn, t.Any args, *t.Any kwargs)

Data Fields
	sleep

	stop

	wait

	retry

	before

	after

	before_sleep

	reraise

	retry_error_cls

	retry_error_callback

Protected Attributes
	_local

Detailed Description

Definition at line 191 of file __init__.py.

Constructor & Destructor Documentation

◆ init()

__init__	(		self,
		t.Callable[[t.Union[int, float]], None]	sleep = `sleep`,
		"StopBaseT"	stop = `stop_never`,
		"WaitBaseT"	wait = `wait_none()`,
		"RetryBaseT"	retry = `retry_if_exception_type()`,
		t.Callable[["RetryCallState"], None]	before = `before_nothing`,
		t.Callable[["RetryCallState"], None]	after = `after_nothing`,
		t.Optional[t.Callable[["RetryCallState"], None]]	before_sleep = `None`,
		bool	reraise = `False`,
		t.Type[RetryError]	retry_error_cls = `RetryError`,
		t.Optional[t.Callable[["RetryCallState"], t.Any]]	retry_error_callback = `None`
	)

Reimplemented in TornadoRetrying, and AsyncRetrying.

Definition at line 192 of file __init__.py.

    ):
        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
 

References i.

Referenced by Protocol.__init_subclass__().

Here is the caller graph for this function:

Member Function Documentation

◆ call()

WrappedFnReturnT __call__	(		self,
		t.Callable[..., WrappedFnReturnT]	fn,
		*t.Any	args,
		**t.Any	kwargs
	)

Reimplemented in TornadoRetrying, Retrying, and AsyncRetrying.

Definition at line 359 of file __init__.py.

    ) -> WrappedFnReturnT:
        pass

References i.

◆ iter()

t.Generator[AttemptManager, None, None] __iter__ ( self )

Reimplemented in AsyncRetrying.

Definition at line 344 of file __init__.py.

    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
 

References Bar.begin, BaseRetrying.begin(), i, BaseRetrying.iter(), RetryAction.sleep, BaseRetrying.sleep, AsyncRetrying.sleep, TornadoRetrying.sleep, and Retry.sleep().

Here is the call graph for this function:

◆ repr()

str __repr__ ( self )

Definition at line 244 of file __init__.py.

    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})>"
        )
 

◆ begin()

None begin ( self )

Definition at line 301 of file __init__.py.

    def begin(self) -> None:
        self.statistics.clear()
        self.statistics["start_time"] = time.monotonic()
        self.statistics["attempt_number"] = 1
        self.statistics["idle_for"] = 0
 

References clear, i, and BaseRetrying.statistics().

Referenced by AsyncRetrying.__aiter__(), TornadoRetrying.__call__(), Retrying.__call__(), AsyncRetrying.__call__(), BaseRetrying.__iter__(), and Bar.__rich_console__().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ copy()

"BaseRetrying" copy	(		self,
		t.Union[t.Callable[[t.Union[int, float]], None], object]	sleep = `_unset`,
		t.Union["StopBaseT", object]	stop = `_unset`,
		t.Union["WaitBaseT", object]	wait = `_unset`,
		t.Union[retry_base, object]	retry = `_unset`,
		t.Union[t.Callable[["RetryCallState"], None], object]	before = `_unset`,
		t.Union[t.Callable[["RetryCallState"], None], object]	after = `_unset`,
		t.Union[t.Optional[t.Callable[["RetryCallState"], None]], object]	before_sleep = `_unset`,
		t.Union[bool, object]	reraise = `_unset`,
		t.Union[t.Type[RetryError], object]	retry_error_cls = `_unset`,
		t.Union[t.Optional[t.Callable[["RetryCallState"], t.Any]], object]	retry_error_callback = `_unset`
	)

Copy this object with some parameters changed if needed.

Definition at line 217 of file __init__.py.

    ) -> "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),
        )
 

References FormatControl.__class__, _InstallRequirementBackedCandidate.__class__, AlreadyInstalledCandidate.__class__, ExtrasCandidate.__class__, OrderedDict.__class__, InfinityType.__class__, NegativeInfinityType.__class__, _IndividualSpecifier.__class__, Distribution.__class__, ParseExpression.__class__, ParamSpec.__class__, _ConcatenateGenericAlias.__class__, _UnpackAlias.__class__, TypeVarTuple.__class__, PoolError.__class__, RequestError.__class__, pip._vendor.tenacity._first_set(), BaseRetrying.after, BaseRetrying.before, BaseRetrying.before_sleep, i, RetryError.reraise(), BaseRetrying.reraise, BaseRetrying.retry, BaseRetrying.retry_error_callback, BaseRetrying.retry_error_cls, RetryAction.sleep, BaseRetrying.sleep, AsyncRetrying.sleep, TornadoRetrying.sleep, Retry.sleep(), Progress.stop(), _RefreshThread.stop(), Live.stop(), Status.stop(), BaseRetrying.stop, and BaseRetrying.wait.

Referenced by ParseResults.__add__(), Text.__add__(), ParserElement.__call__(), ParseResults.__radd__(), ParserElement._setResultsName(), ParseResults.deepcopy(), Text.divide(), ConsoleOptions.reset_height(), Text.split(), ConsoleOptions.update(), ConsoleOptions.update_dimensions(), ConsoleOptions.update_height(), ConsoleOptions.update_width(), Text.with_indent_guides(), and BaseRetrying.wraps().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ iter()

t.Union[DoAttempt, DoSleep, t.Any] iter	(		self,
		"RetryCallState"	retry_state
	)

Definition at line 307 of file __init__.py.

    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 = fut.failed and isinstance(fut.exception(), TryAgain)
        if not (is_explicit_retry or self.retry(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):
            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)
        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)
 

References BaseRetrying.after, BaseRetrying.before, BaseRetrying.before_sleep, i, RetryError.reraise(), BaseRetrying.reraise, BaseRetrying.retry, BaseRetrying.retry_error_callback, BaseRetrying.retry_error_cls, BaseRetrying.statistics(), Progress.stop(), _RefreshThread.stop(), Live.stop(), Status.stop(), BaseRetrying.stop, and BaseRetrying.wait.

Referenced by AsyncRetrying.__anext__(), TornadoRetrying.__call__(), Retrying.__call__(), AsyncRetrying.__call__(), and BaseRetrying.__iter__().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ statistics()

t.Dict[str, t.Any] statistics ( self )

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).

Definition at line 256 of file __init__.py.

    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  # type: ignore[no-any-return]
        except AttributeError:
            self._local.statistics = t.cast(t.Dict[str, t.Any], {})
            return self._local.statistics
 

References BaseRetrying._local, and i.

Referenced by BaseRetrying.begin(), and BaseRetrying.iter().

Here is the caller graph for this function:

◆ wraps()

WrappedFn wraps	(		self,
		WrappedFn	f
	)

Wrap a function for retrying.

:param f: A function to wraps for retrying.

Reimplemented in AsyncRetrying.

Definition at line 283 of file __init__.py.

    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  # type: ignore[attr-defined]
        wrapped_f.retry_with = retry_with  # type: ignore[attr-defined]
 
        return wrapped_f  # type: ignore[return-value]
 