PDF-accessibility-saas/retry_helper.py

#!/usr/bin/env python3
"""
Retry Helper Module

Provides retry logic with exponential backoff for API calls and other operations.
Helps make the application more resilient to transient failures.
"""

import time
import functools
from typing import Callable, Any, Optional, Tuple, Type
from logger_config import setup_logger

logger = setup_logger(__name__, "retry_helper.log")


def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
    """
    Decorator to retry a function with exponential backoff

    Args:
        max_retries: Maximum number of retry attempts (default: 3)
        initial_delay: Initial delay in seconds (default: 1.0)
        max_delay: Maximum delay in seconds (default: 60.0)
        exponential_base: Base for exponential backoff (default: 2.0)
        exceptions: Tuple of exceptions to catch and retry (default: all exceptions)

    Returns:
        Decorated function with retry logic

    Example:
        @retry_with_backoff(max_retries=3, initial_delay=1.0)
        def call_api():
            return api.get_data()

        # Will retry up to 3 times with delays: 1s, 2s, 4s
        result = call_api()
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            delay = initial_delay
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    # Try to execute the function
                    result = func(*args, **kwargs)

                    # If we retried at least once, log success
                    if attempt > 0:
                        logger.info(
                            f"{func.__name__} succeeded on attempt {attempt + 1}/{max_retries + 1}"
                        )

                    return result

                except exceptions as e:
                    last_exception = e

                    # If this was the last attempt, don't retry
                    if attempt >= max_retries:
                        logger.error(
                            f"{func.__name__} failed after {max_retries + 1} attempts: {str(e)}"
                        )
                        raise

                    # Calculate delay with exponential backoff
                    current_delay = min(delay, max_delay)

                    logger.warning(
                        f"{func.__name__} failed on attempt {attempt + 1}/{max_retries + 1}: {str(e)}. "
                        f"Retrying in {current_delay:.1f}s..."
                    )

                    # Wait before retrying
                    time.sleep(current_delay)

                    # Increase delay for next attempt
                    delay *= exponential_base

            # Should never reach here, but just in case
            raise last_exception

        return wrapper
    return decorator


def retry_on_failure(
    func: Callable,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,)
) -> Any:
    """
    Retry a function call with exponential backoff (non-decorator version)

    Args:
        func: Function to execute
        max_retries: Maximum number of retry attempts
        initial_delay: Initial delay in seconds
        exceptions: Tuple of exceptions to catch and retry

    Returns:
        Result of the function call

    Example:
        def api_call():
            return api.get_data()

        result = retry_on_failure(api_call, max_retries=3)
    """
    @retry_with_backoff(max_retries=max_retries, initial_delay=initial_delay, exceptions=exceptions)
    def wrapped():
        return func()

    return wrapped()


class RetryableError(Exception):
    """Exception that indicates an operation should be retried"""
    pass


class NonRetryableError(Exception):
    """Exception that indicates an operation should NOT be retried"""
    pass


def is_retryable_error(error: Exception) -> bool:
    """
    Determine if an error should be retried

    Args:
        error: Exception to check

    Returns:
        True if error should be retried, False otherwise
    """
    # Don't retry explicit non-retryable errors
    if isinstance(error, NonRetryableError):
        return False

    # Retry explicit retryable errors
    if isinstance(error, RetryableError):
        return True

    # Check for common retryable error messages/types
    error_str = str(error).lower()

    retryable_patterns = [
        'timeout',
        'connection',
        'network',
        'temporary',
        'unavailable',
        'rate limit',
        'too many requests',
        '429',
        '503',
        '504',
    ]

    return any(pattern in error_str for pattern in retryable_patterns)


def safe_execute(
    func: Callable,
    fallback_value: Any = None,
    log_errors: bool = True
) -> Any:
    """
    Execute a function and return a fallback value on error (graceful degradation)

    Args:
        func: Function to execute
        fallback_value: Value to return if function fails (default: None)
        log_errors: Whether to log errors (default: True)

    Returns:
        Result of function or fallback value on error

    Example:
        # If API fails, return empty list instead of crashing
        results = safe_execute(
            lambda: api.get_results(),
            fallback_value=[],
            log_errors=True
        )
    """
    try:
        return func()
    except Exception as e:
        if log_errors:
            logger.warning(f"Function {func.__name__} failed gracefully: {str(e)}")
        return fallback_value


if __name__ == "__main__":
    # Test the retry logic
    print("Testing retry_with_backoff decorator...")

    attempt_count = 0

    @retry_with_backoff(max_retries=3, initial_delay=0.5)
    def flaky_function():
        """Simulates a flaky API that fails twice then succeeds"""
        global attempt_count
        attempt_count += 1

        if attempt_count < 3:
            raise ConnectionError(f"Connection failed (attempt {attempt_count})")

        return "Success!"

    try:
        result = flaky_function()
        print(f"✅ Result: {result}")
        print(f"✅ Took {attempt_count} attempts")
    except Exception as e:
        print(f"❌ Failed: {e}")

    # Test safe_execute
    print("\nTesting safe_execute...")

    def failing_function():
        raise ValueError("This always fails")

    result = safe_execute(
        failing_function,
        fallback_value="Fallback value",
        log_errors=True
    )
    print(f"✅ Graceful degradation result: {result}")

    print("\n✅ All tests passed!")