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 RetryError 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: ~tenacity.stop.stop_base | ~typing.Callable[[~tenacity.RetryCallState], bool] = <tenacity.stop.stop_after_attempt object>, wait: ~tenacity.wait.wait_base | ~typing.Callable[[~tenacity.RetryCallState], float | int] = <tenacity.wait.wait_fixed object>, retry: ~tenacity.retry.retry_base | ~typing.Callable[[~tenacity.RetryCallState], bool] = <tenacity.retry.retry_if_exception_type object>, reraise: bool = True, after: ~typing.Callable[[~tenacity.RetryCallState], None] = <function after_nothing>, exc_msg: str | None = None, exc_type: ~typing.Type[BaseException] | None = None, **kwargs)

Bases: object

Handles retry logic for any code block using the Tenacity library.

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

Executes a function with retries, according to the handler’s 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:

RetryError: By default, if all retry attempts fail, encapsulating the original exception.

Optionally, can raise the exception of the last attempt or a specific type and/or value.