Core Decorator & Classes

from tenacity import retry

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.
    Supports both synchronous and asynchronous functions seamlessly.
    
    Parameters:
    - sleep: Function to use for sleeping between retries (default: time.sleep)
    - stop: Strategy determining when to stop retrying (default: never stop)
    - wait: Strategy determining delay between retries (default: no wait)
    - retry: Strategy determining whether to retry after failure (default: retry on any exception)
    - before: Callback executed before each attempt (default: no-op)
    - after: Callback executed after each attempt (default: no-op) 
    - before_sleep: Callback executed before sleeping between retries (default: None)
    - reraise: If True, reraise original exception instead of RetryError (default: False)
    - retry_error_cls: Exception class to raise when retries exhausted (default: RetryError)
    - retry_error_callback: Callback executed when retries exhausted (default: None)
    
    Returns:
    Decorated function with retry behavior applied.
    """

# Basic retry on any exception
@retry
def might_fail():
    pass

# Retry with exponential backoff and attempt limit  
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def api_call():
    pass

# Retry specific exceptions only
@retry(retry=retry_if_exception_type((ConnectionError, TimeoutError)))
def network_operation():
    pass

# Auto-detects async functions
@retry(stop=stop_after_delay(30))
async def async_operation():
    pass

from tenacity import BaseRetrying

class BaseRetrying(ABC):
    """
    Abstract base class for all retry controllers.
    
    Provides the common interface and shared functionality for retry behavior.
    All concrete retry controllers inherit from this class.
    """
    
    def __init__(
        self,
        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
    ):
        """Initialize retry controller with strategies and callbacks."""
    
    def copy(self, **kwargs) -> 'Self':
        """
        Create a copy of this retrying object with modified parameters.
        
        Parameters can be any of the constructor parameters.
        Useful for creating variations of retry behavior.
        """
    
    def wraps(self, f: WrappedFn) -> WrappedFn:
        """
        Wrap a function with this retry behavior.
        
        Returns a new function that will retry according to this controller's configuration.
        """
    
    def begin(self) -> RetryCallState:
        """
        Initialize a new retry session.
        
        Creates and returns a RetryCallState tracking the retry session.
        """
    
    def iter(self, retry_state: RetryCallState) -> Iterator[AttemptManager]:
        """
        Execute one iteration of the retry loop.
        
        Yields AttemptManager context managers for each retry attempt.
        """
    
    @abstractmethod
    def __call__(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
        """Execute function with retry logic. Must be implemented by subclasses."""
    
    def __iter__(self) -> Iterator[AttemptManager]:
        """Iterator interface for attempt managers."""
    
    @property
    def statistics(self) -> dict:
        """Runtime statistics for this retry controller."""
    
    @property  
    def iter_state(self) -> IterState:
        """Current iteration state."""

from tenacity import Retrying

class Retrying(BaseRetrying):
    """
    Standard synchronous retry controller.
    
    Handles retry logic for regular (non-async) functions.
    This is the default controller used by the @retry decorator for sync functions.
    """
    
    def __call__(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
        """
        Execute function with synchronous retry logic.
        
        Parameters:
        - fn: Function to execute with retries
        - *args: Positional arguments to pass to fn
        - **kwargs: Keyword arguments to pass to fn
        
        Returns:
        Result of successful function execution.
        
        Raises:
        RetryError: When all retry attempts are exhausted
        """

from tenacity import AsyncRetrying

class AsyncRetrying(BaseRetrying):
    """
    Asynchronous retry controller for coroutines.
    
    Handles retry logic for async/await functions and coroutines.
    Auto-detects trio vs asyncio for appropriate sleep function.
    """
    
    async def __call__(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
        """
        Execute coroutine with asynchronous retry logic.
        
        Parameters:
        - fn: Async function/coroutine to execute with retries
        - *args: Positional arguments to pass to fn  
        - **kwargs: Keyword arguments to pass to fn
        
        Returns:
        Result of successful coroutine execution.
        
        Raises:
        RetryError: When all retry attempts are exhausted
        """
    
    def __aiter__(self) -> AsyncIterator[AttemptManager]:
        """Async iterator interface for attempt managers."""
    
    async def __anext__(self) -> AttemptManager:
        """Async iteration support."""

from tenacity.tornadoweb import TornadoRetrying

class TornadoRetrying(BaseRetrying):
    """
    Tornado web framework retry controller.
    
    Specialized retry controller for Tornado's @gen.coroutine decorated functions.
    Uses Tornado's IOLoop for asynchronous sleep operations.
    """
    
    @gen.coroutine
    def __call__(self, fn: Callable[..., Any], *args, **kwargs) -> Any:
        """
        Execute Tornado coroutine with retry logic.
        
        Uses Tornado's generator-based coroutine model and IOLoop.sleep.
        
        Parameters:
        - fn: Tornado coroutine to execute with retries
        - *args: Positional arguments to pass to fn
        - **kwargs: Keyword arguments to pass to fn
        
        Returns:
        tornado.concurrent.Future with result of successful execution.
        """

from tenacity import RetryCallState

class RetryCallState:
    """
    Tracks the complete state of a retry session.
    
    Contains all information about the current retry attempt, timing,
    outcomes, and configuration. Passed to all callbacks and strategies.
    """
    
    # Core execution context
    start_time: float              # When the retry session began
    retry_object: BaseRetrying     # The retry controller instance
    fn: Callable[..., Any]         # Function being retried
    args: tuple                    # Function positional arguments  
    kwargs: dict                   # Function keyword arguments
    
    # Attempt tracking
    attempt_number: int            # Current attempt number (starts at 1)
    outcome: Future                # Result of last attempt
    outcome_timestamp: Optional[float]  # When outcome was set
    
    # Timing information
    idle_for: float                # Total time spent sleeping
    upcoming_sleep: float          # Next sleep duration
    seconds_since_start: float     # Elapsed time since first attempt
    
    # Flow control
    next_action: Optional[BaseAction]  # Next action to take (retry/stop)
    
    def prepare_for_next_attempt(self) -> None:
        """Reset state for the next retry attempt."""
    
    def set_result(self, val: Any) -> None:
        """Set successful result for this attempt."""
    
    def set_exception(self, exc_info: tuple) -> None:
        """Set exception result for this attempt."""

from tenacity import Future

class Future:
    """
    Container for attempt results (success or exception).
    
    Encapsulates the outcome of a single retry attempt, including
    the attempt number and whether it failed.
    """
    
    def __init__(self, attempt_number: int):
        """
        Initialize Future for given attempt number.
        
        Parameters:
        - attempt_number: Which attempt this result represents
        """
    
    @property
    def attempt_number(self) -> int:
        """The attempt number this result is from."""
    
    @property  
    def failed(self) -> bool:
        """True if this attempt resulted in an exception."""
    
    @classmethod
    def construct(
        cls, 
        attempt_number: int, 
        value: Any, 
        has_exception: bool
    ) -> 'Future':
        """
        Create Future with result value or exception.
        
        Parameters:
        - attempt_number: Which attempt this represents
        - value: The result value or exception
        - has_exception: True if value is an exception
        
        Returns:
        Future instance with the result set
        """

from tenacity import AttemptManager

class AttemptManager:
    """
    Context manager for individual retry attempts.
    
    Handles entering/exiting attempt context and capturing
    results or exceptions from the attempt.
    """
    
    def __init__(self, retry_state: RetryCallState):
        """Initialize attempt manager with retry state."""
    
    def __enter__(self) -> 'AttemptManager':
        """Enter attempt context."""
    
    def __exit__(
        self, 
        exc_type: Optional[type], 
        exc_value: Optional[BaseException], 
        traceback: Optional[Any]
    ) -> Optional[bool]:
        """
        Exit attempt context and capture result.
        
        Captures successful results or exceptions and updates retry state.
        """

from tenacity import BaseAction

class BaseAction:
    """
    Abstract base class for retry actions.
    
    Represents different actions that can be taken during retry logic.
    """
    
    REPR_FIELDS: tuple = ()  # Fields to include in string representation
    NAME: str = ""           # Human-readable action name

from tenacity import RetryAction

class RetryAction(BaseAction):
    """
    Action representing a retry with specified sleep time.
    
    Indicates that another attempt should be made after sleeping
    for the specified duration.
    """
    
    def __init__(self, sleep: float):
        """
        Initialize retry action with sleep duration.
        
        Parameters:
        - sleep: Number of seconds to sleep before next attempt
        """
    
    @property
    def sleep(self) -> float:
        """Sleep duration for this retry action."""

from tenacity import DoAttempt, DoSleep

class DoAttempt:
    """Marker class indicating an attempt should be made."""
    pass

class DoSleep(float):
    """
    Marker class indicating sleep should occur.
    
    Inherits from float to carry the sleep duration.
    """
    pass

from tenacity import IterState

@dataclass
class IterState:
    """
    Tracks iteration state within the retry loop.
    
    Maintains the current state of retry condition evaluation,
    actions to execute, and timing information.
    """
    
    actions: list = field(default_factory=list)
    retry_run_result: bool = False
    delay_since_first_attempt: float = 0
    stop_run_result: bool = False
    is_explicit_retry: bool = False
    
    def reset(self) -> None:
        """Reset all fields to their default values."""

from tenacity import NO_RESULT, WrappedFn, WrappedFnReturnT

# Sentinel object for unset results
NO_RESULT: object

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

# Create retry controller directly
retrying = Retrying(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)

# Use with different functions
result1 = retrying(function1, arg1, arg2)
result2 = retrying(function2, kwarg=value)

# Create variations
fast_retrying = retrying.copy(wait=wait_fixed(1))

# Async retry controller
async_retrying = AsyncRetrying(
    stop=stop_after_delay(30),
    retry=retry_if_exception_type(ConnectionError)
)

# Use with async functions
result = await async_retrying(async_function, param=value)

# Manual retry loop with attempt managers
retrying = Retrying(stop=stop_after_attempt(3))

for attempt in retrying:
    with attempt:
        # Your code that might fail
        result = risky_operation()
        break  # Success - exit retry loop