functional programming with coroutines python

Question

I have async call, for example

from httpx import AsyncClient, Response

client = AsyncClient()
my_call = client.get(f"{HOST}/api/my_method")  # async call 

And I want to pass it to some retry logic like

async def retry_http(http_call):
    count = 5
    status, response = None, None
    while count > 0:
        response: Response = await http_call
        if response.status_code == 200:
            break
        count -= 1
        if response.status_code in (502, 504):
            await asyncio.sleep(2)
        else:
            break

    if response.status_code != 200:
        return {
            "success": False,
            "result": {
                "error": f"Response Error",
                "response_code": response.status_code,
                "response": response.text,
            }
        }

    return response.json()

await retry_http(my_call)

but I got

RuntimeError
cannot reuse already awaited coroutine

Are there any method to make my_call an reusable coroutine ?

Answer 1

It is not possible in Python - a co-routine, once created, have an internal state that can't be easily duplicated - so once it runs, the internal state changes, including the internal line of code that is in execution, and there is no way to "rewind" that.

The most simple approach is to do like in @RyabchenkoAlexander's answer and accept the co-routine function and its parameters separately, and create the co-routine inside your retry function.

An alternative that is a nice Python idiom is to decorate the co-routine function - you make your retry_http a decorator instead, which wraps the underlying co-routine function in the retrying code.

Then, if the functions where you want this behavior are in your code, you can use the decorator syntax (@name prefixing the function definion) so that all calls will have the retry behavior, or you can apply it as a plain expression to get a new, retriable, co-routine function. Your final call could be:

    result = await (retry_http(client.get) (f"{HOST}/api/my_method"))

(note the extra pair of parentheses around client.get, decorating it)

The decorator itself could be:

def retry_http(coro_func):
    async def wrapper(*args, **kw):
        # your original code - just replacing the await expression
        ...
        while count > 5:
            ...
            result = await coro_func(*args, **kw)
            ...
        ...
        return result
    return wrapper

As for your original intent: it would actually be possible to introspect a coroutine object, its internal variables and passed parameter, to recreate a co-routine object that has not yet started - however, that would involve using introspection to locate the original callable, and making the call again - it would be cumbersome, could be slow, and for little gain. I will outline the requirements, nonetheless:

A co-routine object has the cr_code and cr_frame attributes - you'd need to retrieve the function associated with the code object in cr_code- probably using the garbage colector API, or recreate a new function re-using the same code object, by calling types.FunctionType with the same parameters - and the local and global variables can be retrieved from the frame object in cr_frame.

Answer 2

can be fixed in next way

async def retry_http(http_call, *args, **kwargs):
    count = 5
    status, response = None, None
    while count > 0:
        response: Response = await http_call(*args, **kwargs)
        if response.status_code == 200:
            break
        count -= 1
        if response.status_code in (502, 504):
            await asyncio.sleep(2)
        else:
            break

    if response.status_code != 200:
        return {
            "success": False,
            "result": {
                "error": f"Response Error",
                "response_code": response.status_code,
                "response": response.text,
            }
        }

    return response.json()



client = AsyncClient()
await retry_http(client.get, f"{HOST}/api/my_method")