Tenacity - Comprehensive Python Retry Library

pip install tenacity

import tenacity
from tenacity import retry, stop_after_attempt, wait_exponential

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def unreliable_function():
    """Function that might fail but will be retried up to 3 times with 1 second between attempts."""
    # Your code here
    pass

from typing import Any, Callable, Union, TypeVar, Optional
from datetime import timedelta
import threading

# Type aliases for time units
time_unit_type = Union[int, float, timedelta]

# Type variables for wrapped functions
WrappedFn = TypeVar('WrappedFn', bound=Callable[..., Any])
WrappedFnReturnT = TypeVar('WrappedFnReturnT')

# Base type aliases for strategy composition
RetryBaseT = Union['retry_base', Callable[['RetryCallState'], bool]]
StopBaseT = Union['stop_base', Callable[['RetryCallState'], bool]]  
WaitBaseT = Union['wait_base', Callable[['RetryCallState'], Union[float, int]]]

# Core state management
class RetryCallState:
    """Tracks state of a retry session."""
    start_time: float
    retry_object: 'BaseRetrying'
    fn: Callable[..., Any]
    args: tuple
    kwargs: dict
    attempt_number: int
    outcome: 'Future'
    outcome_timestamp: Optional[float]
    idle_for: float
    next_action: Optional['BaseAction']
    upcoming_sleep: float
    seconds_since_start: float

class Future:
    """Container for attempt results."""
    attempt_number: int
    failed: bool
    
    @classmethod
    def construct(cls, attempt_number: int, value: Any, has_exception: bool) -> 'Future'

# Exception classes
class TryAgain(Exception):
    """Raise to force immediate retry regardless of retry condition."""
    pass

class RetryError(Exception):
    """Raised when all retry attempts are exhausted."""
    last_attempt: Future
    
    def reraise(self) -> None:
        """Reraise the original exception from the last attempt."""

def retry(
    sleep: Callable[[float], None] = sleep,
    stop: StopBaseT = stop_never,
    wait: WaitBaseT = wait_none(),
    retry: RetryBaseT = retry_if_exception_type(),
    before: Callable[[RetryCallState], None] = before_nothing,
    after: Callable[[RetryCallState], None] = after_nothing,
    before_sleep: Optional[Callable[[RetryCallState], None]] = None,
    reraise: bool = False,
    retry_error_cls: type = RetryError,
    retry_error_callback: Optional[Callable[[RetryCallState], Any]] = None
) -> Callable[[WrappedFn], WrappedFn]:
    """
    Main decorator to add retry behavior to functions.
    
    Auto-detects async/tornado functions and applies appropriate retry controller.
    
    Parameters:
    - sleep: Function to use for sleeping between retries
    - stop: Strategy determining when to stop retrying
    - wait: Strategy determining how long to wait between retries  
    - retry: Strategy determining whether to retry after a failure
    - before: Callback executed before each attempt
    - after: Callback executed after each attempt
    - before_sleep: Callback executed before sleeping between retries
    - reraise: If True, reraise original exception instead of RetryError
    - retry_error_cls: Exception class to raise when retries are exhausted
    - retry_error_callback: Callback executed when retries are exhausted
    """

# Exponential backoff with jitter for API calls
@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(min=1, max=60),
    retry=retry_if_exception_type((ConnectionError, TimeoutError))
)
def call_external_api():
    pass

# Database connection with custom retry logic  
@retry(
    stop=stop_after_delay(30),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(ConnectionError),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def connect_to_database():
    pass

# Async function with comprehensive retry
@retry(
    stop=stop_any(stop_after_attempt(5), stop_after_delay(60)),
    wait=wait_combine(wait_fixed(3), wait_random(0, 2)),
    retry=retry_if_not_result(lambda x: x is not None)
)
async def fetch_data():
    pass