How to assign yield value to variable

Question

I was trying to create a dynamic decorator function for both sync and async function and it is working fine, but I cannot use yield value further.

def retry(f) -> Any:
    @contextmanager
    def retry_context(*args):
        # response = False
        i = 0
        response_ok = False

        retries = DEFAULT_RETRIES
        status_retries = DEFAULT_STATUS_RETRIES

        retry_count = 0
        while i < retries and response_ok is False:
            retry_count = retry_count+1
            yield response
            i += 1
            if response.status_code not in status_retries:
                response_ok = True
                logger.debug("Response {}".format(response))
        if not response_ok:
            logger.warning("Response ERROR: {}".format(response))
        return response


    def wrapper(*args, **kwargs) -> Any:
        with retry_context(*args):
            return f(*args, **kwargs)

    async def async_wrapper(*args, **kwargs) -> Any:
        with retry_context(*args):
            return await f(*args, **kwargs)

    if asyncio.iscoroutinefunction(f):
        return async_wrapper
    return wrapper

I have seen many other answers for assigning value to yield but none is solving my issue, as I also want to use it further. Any other solution is also welcome.

Answer 1

The Context Manager protocol (i.e., the with statement ant its associated special methods and helper functions in the contextlib module), or "async context manager" as well, in this case - is not designed and cannot be made to work as a "retry" mechanism by itself.

That means: the context manager, either a generator decorated with the contextlib.contextmanager decorator, or a full-fledged class with an __enter__ and __exit__ methods will only, and can only by the way the language works, pass control to the body of the with statement a single time. For a contextmanager decorated generator, that means it must run a yield expression a single time in its lifetime.

You should either use an outer while loop with a differently designed class, or a ready-made retry library to achieve your goals there.

As for a nice 3rd party library implementing retries with a lot of options and care for edgecases, you can try "tenacity": https://github.com/jd/tenacity

Answer 2

I got it working with tenacity. Thanks everyone for guiding.

from tenacity import retry, retry_if_result, stop_after_attempt

retries = 3  
status_retries = (500, 502)

def is_retryable_status_code(response):
    global status_retries
    return response.status_code in status_retries

def retry_with_status(f) -> Any:
    global retries
    @retry(stop = stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
    def retry_wrapper(*args, **kwargs):
        return f(*args, **kwargs)

    @retry(stop = stop_after_attempt(retries), retry=retry_if_result(is_retryable_status_code))
    async def async_retry_wrapper(*args, **kwargs):
        return await f(*args, **kwargs)

    def wrapper(*args, **kwargs) -> Any:
        global retries, status_retries
        if 'retry' in dir(args[0]) and 'status_retries' in dir(args[0]):
            retries = args[0].retries
            status_retries = args[0].status_retries

        if asyncio.iscoroutinefunction(f):
            return async_retry_wrapper(*args, **kwargs)
        else:
            return retry_wrapper(*args, **kwargs)
    return wrapper