asyncio.run not working at python azure function app

Question

I have a code that works in local perfectly:

... main.py
from file import my_test 
... test.py
async test():
    """some async function"""
    await asyncio.sleep(1)
    return "test"

import asyncio
my_test = asyncio.run(test()) 

Notice that test is called at module level, But in an azure function app this simple code it doesn't work.

raise RuntimeError('This event loop is already running')

Which as far as I know, it means that the function app is already running his own event loop so you can't run it twice.

So i tried to develop a custom run function that captures the event loop and awaits the value if exist, otherwise uses the plain asyncio.run.

But can't figure out how to make it properly, because it needs to work both in local and in azure, and it can't be async and not async at the same time.

def my_run(coro: Coroutine[Any, Any, T], *args, **kwargs) -> T:
    """ Azure uses already asyncio.run and it cannot be nested, so if run in azure we take their event loop"""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro,  *args, **kwargs)
    else:
        return loop.run_until_complete(coro(*args, **kwargs))
my_test = my_run(test())  # far from working

Any idea?

thanks in advance

Answer 1

The code you provided is not working in an Azure function app because Azure function host is already running its own event loop and you cannot run another event loop inside of an Azure function.

To fix this, you can use the my_run() function that you defined This function checks to see if the asyncio event loop is already running. If it is, then the function uses the existing event loop to run the test() function. Otherwise, the function starts a new event loop and runs the test() function.

I have tried from my end , below is the httptrigger function code at module level defined async def test() outside of function i have runned sucessfully in both local and azure function app

import azure.functions as func
import logging
import asyncio
from typing import Any, Coroutine

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

async def test():
    """some async function"""
    await asyncio.sleep(1)
    return "Application Run Successfully"

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request')

    def my_run(coro: Coroutine[Any, Any, Any], *args, **kwargs) -> Any:
        """Azure uses already asyncio.run and it cannot be nested, so if run in Azure, we take their event loop"""
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(coro, *args, **kwargs)
        else:
            return loop.run_until_complete(coro(*args, **kwargs))

    result = my_run(test())  # Call your async function using my_run
    print(result)
    return func.HttpResponse(result)

Local Run

Deployed in Azure Function app