Decorators

Retry

This module provides a flexible retry mechanism based on the tenacity library, allowing for advanced retry strategies with exponential backoff, custom exception handling, and detailed logging.

from core_extensions.decorators.retry import SimpleRetry

# Basic usage with default settings
retry_handler = SimpleRetry(max_attempts=3, base_delay=1.0)

@retry_handler.create_decorator((ConnectionError, TimeoutError))
def fetch_data():
    # Function that may fail transiently
    return api.get_data()

# Advanced usage with custom logger and settings
import logging

custom_logger = logging.getLogger(__name__)
retry_handler = SimpleRetry(
    max_attempts=5,
    base_delay=2.0,
    max_delay=30.0,
    reraise=False,
    logger=custom_logger
)

@retry_handler.create_decorator((ValueError, RuntimeError), reraise=True)
def process_data():
    # Function with custom retry behavior
    pass

class core_extensions.decorators.retry.SimpleRetry(max_attempts: int = 3, base_delay: float = 1.0, max_delay: float = 10.0, reraise: bool = True, logger: Logger | None = None, **kwargs: Any)

Bases: object

Simple and generic retry mechanism based on tenacity library in case an advanced retry mechanism is required beyond simple capabilities offered by core_mixins.decorators.retry.

__init__(max_attempts: int = 3, base_delay: float = 1.0, max_delay: float = 10.0, reraise: bool = True, logger: Logger | None = None, **kwargs: Any) → None

Initialize the SimpleRetry instance.

Parameters:

• max_attempts – Maximum number of retry attempts. Defaults to 3.

• base_delay – Base delay in seconds for exponential backoff. Defaults to 1.0.

• max_delay – Maximum delay in seconds between retries. Defaults to 10.0.

• reraise – Whether to reraise the exception after all retry attempts are exhausted. Defaults to True. If False, tenacity.RetryError is raised.

• logger – Optional custom logger instance. If None, a default logger will be created.

• kwargs – Additional keyword arguments to pass to tenacity.retry decorator.

after_error(ref: RetryCallState)

Callback executed after each failed attempt. It logs the exception type that caused the failure using info level to prevent error notifications while retrying.

Parameters:

ref – RetryCallState containing information about the retry attempt.

before_sleep(ref: RetryCallState)

Callback executed before sleeping between retry attempts. It logs the retry attempt number, function name, and the upcoming wait duration before the next attempt.

Parameters:

ref – RetryCallState containing information about the retry attempt.

create_decorator(exception_types: Tuple[Type[Exception], ...], reraise: bool | None = None) → Callable

Create a retry decorator for specified exception types.

Parameters:

• exception_types – Tuple of exception types to catch and retry on. Must contain at least one exception type.

• reraise – Optional override for whether to reraise exceptions after all attempts. If None, uses the instance’s reraise setting. If False, tenacity.RetryError is raised.

Returns:

A tenacity retry decorator configured with the instance’s settings.

Raises:

ValueError – If exception_types is empty.