waft.utilities.retry_handler module

This module provides a RetryHandler class that encapsulates retry logic using the tenacity library. The class allows for flexible and robust error handling by retrying operations under specific conditions, with customizable retry configurations.

The RetryHandler can be used to apply retry mechanisms to any callable function or method, making it ideal for operations that may fail due to transient issues such as network requests, file operations, or interactions with external services.

Example Usage:

def some_function(param: int) -> str:

print(f”Trying with {param}”) if param < 3:

raise ValueError(“Parameter is too low”)

return f”Success with {param}”

retry_handler = RetryHandler() result = retry_handler.execute(some_function, 3) print(result)

Custom Configuration Examples:

1. Adjusting Stop Conditions:

   # Stop after 5 attempts retry_handler = RetryHandler(stop=stop_after_attempt(5))

   # Stop after 30 seconds retry_handler = RetryHandler(stop=stop_after_delay(30))

2. Modifying Wait Strategies:

   # Fixed wait of 5 seconds between retries retry_handler = RetryHandler(wait=wait_fixed(5))

   # Random wait between 1 and 3 seconds retry_handler = RetryHandler(wait=wait_random(1, 3))

3. Specifying Retry Conditions:

   # Retry only for ValueError exceptions retry_handler = RetryHandler(retry=retry_if_exception_type(ValueError))

   # Retry when the function returns False retry_handler = RetryHandler(retry=retry_if_result(lambda result: result is False))

4. Controlling Exception Re-raising:

   # Do not re-raise exceptions, handle differently retry_handler = RetryHandler(reraise=False) try:

   result = retry_handler.execute(some_function, 1)

   except RuntimeError as e:

   print(“Handling failure gracefully”)

Attributes:

retry_config (dict): Configuration for retry operations, which can include keys like ‘stop’, ‘wait’, ‘retry’, and ‘re-raise’.

Classes:

RetryHandler: Handles retry logic for callable functions using customizable retry configurations.

Functions:

some_function(param: int) -> str: Example function to demonstrate how to use RetryHandler.

class waft.utilities.retry_handler.RetryHandler(stop: ~typing.Any | None = None, wait: ~typing.Any | None = None, retry: ~typing.Any | None = None, reraise: bool = True, after: ~typing.Callable[[~tenacity.RetryCallState], None] | None = None, exc_msg: str = 'Operation failed after retries', exc_type: ~typing.Type[BaseException] = <class 'RuntimeError'>)

Bases: object

Handles retry logic for any callable using the tenacity library.

Attributes:

stop (Stop): The condition that determines when to stop retrying. wait (Wait): The strategy to use for waiting between retries. retry (Retry): The condition that determines if a retry should be attempted. reraise (bool): Whether to re-raise the exception after all retries fail. after (callable): Callable to run after any failed attempt.

Must accept RetryCallState as 1st positional argument

exc_msg: Custom message to use in exception raised upon failure exc_type: Type of exception to raise upon retry failure (Default RuntimeError)

execute(func: Callable[[...], Any], *args: Any, **kwargs: Any) → Any

Executes a function with retries configured according to the class’s retry configuration.

Args:

func (Callable[…, Any]): The function to retry. *args: Positional arguments to pass to the function. **kwargs: Keyword arguments to pass to the function.

Returns:

Any: The result of the function if successful.

Raises:

RuntimeError: If all retry attempts fail, encapsulating the original RetryError.