capcat.core.retry
File: Application/capcat/core/retry.py
Description
Retry mechanisms with exponential backoff for Capcat. Provides robust error recovery for network and other transient failures.
Classes
RetryableOperation
Context manager for retryable operations with custom logic.
Methods
init
def __init__(self, operation_name: str, max_retries: int = None)
Parameters:
selfoperation_name(str)max_retries(int) optional
enter
def __enter__(self)
Parameters:
self
exit
def __exit__(self, exc_type, exc_val, exc_tb)
Parameters:
selfexc_typeexc_valexc_tb
should_retry
def should_retry(self, exception: Exception) -> bool
Determine if an exception should trigger a retry.
Args: exception: The exception that occurred
Returns: True if the operation should be retried
Parameters:
selfexception(Exception)
Returns: bool
Functions
exponential_backoff_retry
def exponential_backoff_retry(max_retries: int = None, base_delay: float = None, max_delay: float = 60.0, exponential_base: float = 2.0, jitter: bool = True, retryable_exceptions: Tuple[Type[Exception], ...] = (Exception,), skip_after: int = None)
Decorator that implements exponential backoff retry logic.
Args: max_retries: Maximum number of retry attempts (uses config default if None) base_delay: Base delay in seconds (uses config default if None) max_delay: Maximum delay between retries in seconds exponential_base: Base for exponential backoff calculation jitter: Whether to add random jitter to delay retryable_exceptions: Tuple of exception types that should trigger retry skip_after: Number of attempts after which to skip instead of raising exception (None = never skip)
Parameters:
max_retries(int) optionalbase_delay(float) optionalmax_delay(float) optionalexponential_base(float) optionaljitter(bool) optionalretryable_exceptions(Tuple[Type[Exception], …]) optionalskip_after(int) optional
network_retry
def network_retry(func: Callable) -> Callable
Convenience decorator for network operations with appropriate retry settings. Handles connection errors, timeouts, and transient HTTP errors.
Parameters:
func(Callable)
Returns: Callable
fast_media_retry
def fast_media_retry(func: Callable) -> Callable
Fast retry decorator optimized for media downloads (images, audio, video). Uses shorter delays and fewer retries for better performance with bulk downloads.
Parameters:
func(Callable)
Returns: Callable
decorator
def decorator(func: Callable) -> Callable
Parameters:
func(Callable)
Returns: Callable
wrapper
def wrapper() -> Any
Returns: Any