Retry Strategies

from tenacity.retry import retry_base

class retry_base(ABC):
    """
    Abstract base class for all retry strategies.
    
    Provides logical operators for combining retry conditions
    and defines the interface all retry strategies must implement.
    """
    
    @abstractmethod
    def __call__(self, retry_state: RetryCallState) -> bool:
        """
        Determine whether to retry based on current state.
        
        Parameters:
        - retry_state: Complete state of current retry session
        
        Returns:
        True if another attempt should be made, False to stop retrying
        """
    
    def __and__(self, other: 'retry_base') -> 'retry_all':
        """Combine with another condition using AND logic."""
    
    def __or__(self, other: 'retry_base') -> 'retry_any':
        """Combine with another condition using OR logic."""

from tenacity.retry import RetryBaseT

RetryBaseT = Union[retry_base, Callable[[RetryCallState], bool]]

from tenacity import retry_always, retry_never

# Singleton instances - always retry or never retry
retry_always: retry_base  # Always returns True
retry_never: retry_base   # Always returns False

# Never retry (useful for testing)
@retry(retry=retry_never)
def no_retries():
    pass

# Always retry (combine with stop condition to avoid infinite loops)  
@retry(retry=retry_always, stop=stop_after_attempt(3))
def always_retry_but_limited():
    pass

from tenacity import retry_if_exception

class retry_if_exception(retry_base):
    """
    Retry if exception matches a custom predicate function.
    
    Most flexible exception-based retry strategy - allows complex
    exception analysis beyond just type checking.
    """
    
    def __init__(self, predicate: Callable[[BaseException], bool]):
        """
        Initialize with exception predicate.
        
        Parameters:
        - predicate: Function that takes an exception and returns bool
        """

from tenacity import retry_if_exception_type

class retry_if_exception_type(retry_if_exception):
    """
    Retry if exception is an instance of specified types.
    
    Most commonly used retry strategy - retries on specific exception types.
    """
    
    def __init__(
        self, 
        exception_types: Union[type, tuple[type, ...]] = Exception
    ):
        """
        Initialize with exception types to retry on.
        
        Parameters:
        - exception_types: Single exception type or tuple of types to retry on
                          Defaults to Exception (retry on any exception)
        """

from tenacity import retry_if_not_exception_type

class retry_if_not_exception_type(retry_if_exception):
    """
    Retry unless exception is an instance of specified types.
    
    Inverse of retry_if_exception_type - retries on all exceptions
    except the specified types.
    """
    
    def __init__(
        self, 
        exception_types: Union[type, tuple[type, ...]] = Exception
    ):
        """
        Initialize with exception types to NOT retry on.
        
        Parameters:
        - exception_types: Exception types that should stop retrying
        """

from tenacity import retry_unless_exception_type

class retry_unless_exception_type(retry_if_exception):
    """
    Retry until exception of specified type is raised.
    
    Always retries on success, only stops when the specified
    exception type is encountered. Useful for "retry until error" patterns.
    """
    
    def __init__(
        self, 
        exception_types: Union[type, tuple[type, ...]] = Exception
    ):
        """
        Initialize with exception types that should stop retrying.
        
        Parameters:
        - exception_types: Exception types that indicate completion
        """

from tenacity import retry_if_exception_cause_type

class retry_if_exception_cause_type(retry_base):
    """
    Retry if any exception in the cause chain matches specified types.
    
    Examines the entire exception cause chain (exception.__cause__ and
    exception.__context__) to find matching exception types.
    """
    
    def __init__(
        self, 
        exception_types: Union[type, tuple[type, ...]] = Exception
    ):
        """
        Initialize with exception types to search for in cause chain.
        
        Parameters:  
        - exception_types: Types to look for in exception cause chain
        """

from tenacity import retry_if_exception_message, retry_if_not_exception_message
import re
from typing import Pattern

class retry_if_exception_message(retry_if_exception):
    """
    Retry if exception message equals string or matches regex pattern.
    
    Provides fine-grained control based on exception message content.
    """
    
    def __init__(
        self, 
        message: Optional[str] = None,
        match: Optional[Union[str, Pattern[str]]] = None
    ):
        """
        Initialize with message matching criteria.
        
        Parameters:
        - message: Exact message string to match (mutually exclusive with match)
        - match: Regex pattern to match against message (mutually exclusive with message)
        
        Note: Exactly one of message or match must be provided.
        """

class retry_if_not_exception_message(retry_if_exception_message):
    """
    Retry until exception message equals string or matches regex pattern.
    
    Always retries on success, only stops when exception message matches.
    """

from tenacity import retry_if_result

class retry_if_result(retry_base):
    """
    Retry if successful result matches a predicate function.
    
    Allows retrying based on return values rather than exceptions.
    Useful for APIs that return error codes or sentinel values.
    """
    
    def __init__(self, predicate: Callable[[Any], bool]):
        """
        Initialize with result predicate.
        
        Parameters:
        - predicate: Function that takes return value and returns bool
                    True means retry, False means accept result
        """

from tenacity import retry_if_not_result

class retry_if_not_result(retry_base):
    """
    Retry if successful result does NOT match a predicate function.
    
    Inverse of retry_if_result - retries until the predicate returns True.
    """
    
    def __init__(self, predicate: Callable[[Any], bool]):
        """
        Initialize with result predicate.
        
        Parameters:
        - predicate: Function that takes return value and returns bool
                    False means retry, True means accept result
        """

from tenacity import retry_any

class retry_any(retry_base):
    """
    Retry if ANY of the provided conditions are true (OR logic).
    
    Combines multiple retry strategies with logical OR.
    """
    
    def __init__(self, *retries: retry_base):
        """
        Initialize with retry strategies to combine.
        
        Parameters:
        - *retries: Variable number of retry strategies
        
        Returns True if any strategy returns True.
        """

from tenacity import retry_all

class retry_all(retry_base):
    """
    Retry if ALL of the provided conditions are true (AND logic).
    
    Combines multiple retry strategies with logical AND.
    """
    
    def __init__(self, *retries: retry_base):
        """
        Initialize with retry strategies to combine.
        
        Parameters:
        - *retries: Variable number of retry strategies
        
        Returns True only if all strategies return True.
        """

# Retry on specific exception types
@retry(retry=retry_if_exception_type((ConnectionError, TimeoutError)))
def network_call():
    pass

# Retry on any exception except KeyboardInterrupt
@retry(retry=retry_if_not_exception_type(KeyboardInterrupt))
def interruptible_operation():
    pass

# Retry until ValueError is raised (success or other exceptions continue)
@retry(retry=retry_unless_exception_type(ValueError))
def retry_until_validation_error():
    pass

# Retry on specific error message
@retry(retry=retry_if_exception_message(message="Connection timeout"))
def api_with_timeout():
    pass

# Retry on messages matching regex pattern
@retry(retry=retry_if_exception_message(match=r"rate limit.*exceeded"))
def rate_limited_api():
    pass

# Retry until specific error message appears
@retry(retry=retry_if_not_exception_message(match=r"authentication.*failed"))
def retry_until_auth_fails():
    pass

# Retry if result is None
@retry(retry=retry_if_result(lambda x: x is None))
def get_data():
    pass

# Retry until result meets criteria  
@retry(retry=retry_if_not_result(lambda x: x > 0))
def get_positive_number():
    pass

# Retry on specific return values
@retry(retry=retry_if_result(lambda x: x in ['PENDING', 'PROCESSING']))
def check_job_status():
    pass

# Custom exception analysis
def is_retryable_error(exc):
    return (
        isinstance(exc, ConnectionError) or 
        (isinstance(exc, ValueError) and "temporary" in str(exc))
    )

@retry(retry=retry_if_exception(is_retryable_error))
def complex_operation():
    pass

# Custom result analysis  
def is_incomplete_result(result):
    return isinstance(result, dict) and result.get('status') == 'incomplete'

@retry(retry=retry_if_result(is_incomplete_result))  
def fetch_complete_data():
    pass

# Retry on connection errors OR timeout errors
@retry(retry=retry_any(
    retry_if_exception_type(ConnectionError),
    retry_if_exception_type(TimeoutError)
))
def network_operation():
    pass

# Retry if BOTH conditions are met
@retry(retry=retry_all(
    retry_if_exception_type(ValueError),
    retry_if_exception_message(match=r"retry.*possible")
))
def strict_retry_conditions():
    pass

# Complex nested logic
@retry(retry=retry_any(
    retry_if_exception_type(ConnectionError),
    retry_all(
        retry_if_exception_type(ValueError), 
        retry_if_exception_message(message="Temporary failure")
    ),
    retry_if_result(lambda x: x is None)
))
def complex_retry_logic():
    pass

# Use & and | operators for logical combinations
connection_errors = retry_if_exception_type((ConnectionError, TimeoutError))
temp_failures = retry_if_exception_message(match=r"temp.*fail")

# AND logic using & operator  
strict_retry = connection_errors & temp_failures

# OR logic using | operator
lenient_retry = connection_errors | temp_failures

@retry(retry=lenient_retry)
def flexible_operation():
    pass

# Retry if any exception in cause chain is a ConnectionError
@retry(retry=retry_if_exception_cause_type(ConnectionError))
def wrapped_network_call():
    try:
        network_operation()
    except ConnectionError as e:
        raise ValueError("Network failed") from e

# Retry different exceptions with different conditions
http_errors = retry_if_exception_type((requests.ConnectionError, requests.Timeout))
server_errors = retry_if_exception_message(match=r"5\d\d.*error")
temp_results = retry_if_result(lambda x: x.get('status') == 'temporary')

comprehensive_retry = http_errors | server_errors | temp_results

@retry(
    retry=comprehensive_retry,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def robust_api_call():
    pass

from tenacity import TryAgain

class TryAgain(Exception):
    """
    Exception to force immediate retry regardless of retry condition.
    
    Raise this exception to bypass all retry strategy evaluation
    and force another attempt (subject to stop conditions).
    """
    pass

# Usage example
@retry(stop=stop_after_attempt(3))
def conditional_retry():
    if some_condition():
        raise TryAgain  # Forces retry even if retry strategy would say no
    # Normal logic here