Can I dynamically change the arguments passed to a function when retrying with Tenacity?

Question

I'd like to make use of the Tenacity Python library for its @retry decorator. However, I want to call my function with different parameters on each retry, and am not sure how to specify that.

My function definition looks something like this:

from tenacity import retry, retry_if_exception_type, stop_after_attempt

class CustomError(Exception):
    pass

@retry(retry=retry_if_exception_type(CustomError), stop=stop_after_attempt(2))
def my_function(my_param):
    result = do_some_business_logic(my_param)
    if not result:
        if my_param == 1:
            raise CustomError()
        else:
            raise ValueError()

# first invoke the function with my_param=1, then retry with my_param=2 
my_function(1)

This is a little bit simplified, but the idea is that when I first invoke the function, I'm going to pass in 1 as the first parameter. On retry, I want it to change this value to 2. Can this be done with Tenacity's @retry decorator? Perhaps through a callback?

Answer 1

The easiest way to do this may be to pass in, not an integer, but an iterable object that yields the values you want. For example:

@retry(retry=retry_if_exception_type(CustomError), stop=stop_after_attempt(2))
def my_function(my_iter):
    my_param = next(my_iter)
    result = do_some_business_logic(my_param)
    if not result:
        if my_param == 1:
            raise CustomError()
        else:
            raise ValueError()

my_function(iter([1, 2]))

This does look like an XY problem, though; there is probably a better way to use Tenacity to do what you want to do. Maybe you should post a more general question about retrying.

Answer 2

You can modify arguments in the retry decorator if you call argument with name like "my_param=1".

def change_parameter(new_param):
    def _set_parameter(retry_state):
        retry_state.kwargs['my_param'] = new_param
    return _set_parameter

@retry(
    retry=retry_if_exception_type(CustomError),
    stop=stop_after_attempt(2),
    after=change_parameter(2))
def my_function(my_param):
    print(my_param)
    result = do_some_business_logic(my_param)
    if not result:
        if my_param == 1:
            raise CustomError()
        else:
            raise ValueError()

# first invoke the function with my_param=1, then retry with my_param=2 
my_function(my_param=1)