Retry

Defines retry and timeout utilities.

Authors

Henry Wicaksono (henry.wicaksono@gdplabs.id)

References

NONE

RetryConfig

Bases: BaseModel

Configuration for retry behavior.

Attributes:

Name	Type	Description
max_retries	int	Maximum number of retry attempts.
base_delay	float	Base delay in seconds between retries.
max_delay	float	Maximum delay in seconds between retries.
jitter	bool	Whether to add random jitter to delays.
timeout	float | None	Overall timeout in seconds for the entire operation. If None, timeout is disabled.
retry_on_exceptions	tuple[type[Exception], ...]	Tuple of exception types to retry on.

validate_delay_constraints()

Validates that max_delay is greater than or equal to base_delay.

Returns:

Name	Type	Description
RetryConfig	RetryConfig	The validated configuration.

Raises:

Type	Description
ValueError	If max_delay is less than base_delay.

retry(func_or_config=None, *args, retry_config=None, **kwargs)

Executes a function with retry logic or creates a retry decorator.

This function supports two usage patterns: 1. Direct function execution: await retry(func, args, retry_config=config, *kwargs) 2. Decorator factory: @retry() or @retry(config)

Parameters:

Name	Type	Description	Default
func_or_config	Callable[..., Any] | RetryConfig | None	Either a function to execute or a RetryConfig for decorator usage. Defaults to None.	None
*args	Any	Positional arguments (only used in direct execution mode).	()
retry_config	RetryConfig | None	Retry configuration (only used in direct execution mode). Defaults to None, in which case no retry nor timeout is applied.	None
**kwargs	Any	Keyword arguments (only used in direct execution mode).	{}

Returns:

Type	Description
T | Callable[[Callable[..., Any]], Callable[..., Any]]	T | Callable[[Callable[..., Any]], Callable[..., Any]]: Either the result of function execution or a decorator function.

Raises:

Type	Description
Exception	The last exception raised by the function if all retries are exhausted.
TimeoutError	If the overall timeout is exceeded.

Examples:

Direct function execution

result = await retry(my_function, arg1, arg2, retry_config=config)

Decorator usage - parameterless

@retry()
async def my_async_function():
    # Use default settings, in which case no retry nor timeout is applied.
    pass

Decorator usage - with custom configuration

@retry(RetryConfig(max_retries=3, timeout=120))
async def my_function():
    # Will retry up to 3 times with 0.5s base delay
    pass

Decorator on sync functions

@retry()
def my_sync_function():
    # Works with sync functions too
    return "success"

Decorator on class methods

class MyService:
    @retry(RetryConfig(max_retries=2))
    async def get_data(self, id: str):
        return {"id": id, "data": "value"}