forked from HAL9000/cleveragents-core
100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
"""Type stubs for tenacity module."""
|
|
|
|
from collections.abc import Callable
|
|
from types import TracebackType
|
|
from typing import Any, TypeVar
|
|
|
|
_T = TypeVar("_T")
|
|
|
|
class RetryCallState:
|
|
"""State of a retry call."""
|
|
|
|
attempt_number: int
|
|
outcome: Any | None
|
|
next_action: Any | None
|
|
|
|
class AttemptManager:
|
|
"""Context manager for retry attempts."""
|
|
|
|
retry_state: RetryCallState
|
|
def __enter__(self) -> None: ...
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc_val: BaseException | None,
|
|
exc_tb: TracebackType | None,
|
|
) -> bool: ...
|
|
|
|
class Retrying:
|
|
"""Sync retrying executor."""
|
|
def __init__(
|
|
self,
|
|
stop: Any = None,
|
|
wait: Any = None,
|
|
retry: Any = None,
|
|
before: Callable[[RetryCallState], None] | None = None,
|
|
after: Callable[[RetryCallState], None] | None = None,
|
|
reraise: bool = False,
|
|
) -> None: ...
|
|
def __iter__(self) -> Retrying: ...
|
|
def __next__(self) -> AttemptManager: ...
|
|
|
|
class AsyncRetrying:
|
|
"""Async retrying executor."""
|
|
def __init__(
|
|
self,
|
|
stop: Any = None,
|
|
wait: Any = None,
|
|
retry: Any = None,
|
|
before: Callable[[RetryCallState], None] | None = None,
|
|
after: Callable[[RetryCallState], None] | None = None,
|
|
reraise: bool = False,
|
|
) -> None: ...
|
|
def __aiter__(self) -> AsyncRetrying: ...
|
|
async def __anext__(self) -> AttemptManager: ...
|
|
|
|
def retry(
|
|
stop: Any = None,
|
|
wait: Any = None,
|
|
retry: Any = None,
|
|
before: Callable[[RetryCallState], None] | None = None,
|
|
after: Callable[[RetryCallState], None] | None = None,
|
|
reraise: bool = False,
|
|
) -> Callable[[Callable[..., _T]], Callable[..., _T]]: ...
|
|
def retry_if_exception_type(
|
|
exception_types: type[BaseException] | tuple[type[BaseException], ...],
|
|
) -> Any: ...
|
|
def retry_if_result(predicate: Callable[[Any], bool]) -> Any: ...
|
|
def stop_after_attempt(max_attempt_number: int) -> Any: ...
|
|
def stop_after_delay(max_delay: float) -> Any: ...
|
|
def wait_exponential(
|
|
multiplier: float = 1,
|
|
max: float = 60,
|
|
exp_base: float = 2,
|
|
min: float = 0,
|
|
) -> Any: ...
|
|
def wait_exponential_jitter(
|
|
initial: float = 1,
|
|
max: float = 60,
|
|
exp_base: float = 2,
|
|
jitter: float = 1,
|
|
) -> Any: ...
|
|
def wait_fixed(wait: float) -> Any: ...
|
|
def wait_random(min: float = 0, max: float = 1) -> Any: ...
|
|
|
|
__all__ = [
|
|
"AsyncRetrying",
|
|
"AttemptManager",
|
|
"RetryCallState",
|
|
"Retrying",
|
|
"retry",
|
|
"retry_if_exception_type",
|
|
"retry_if_result",
|
|
"stop_after_attempt",
|
|
"stop_after_delay",
|
|
"wait_exponential",
|
|
"wait_exponential_jitter",
|
|
"wait_fixed",
|
|
"wait_random",
|
|
]
|