Python Azure函数中asyncio.run无法运行的问题咨询

问题根源 #

Azure函数应用的运行时会提前启动自身的事件循环，而你在模块级别直接调用asyncio.run()时，会尝试创建新的事件循环，导致嵌套循环冲突，触发RuntimeError('This event loop is already running')错误。本地环境中模块加载时无运行中的循环，所以asyncio.run()可以正常执行。

解决方案 #

方案一：调整代码结构（推荐） #

避免在模块级别执行异步初始化逻辑，将异步代码移至Azure函数的处理函数内部执行，利用函数运行时已有的事件循环：

修改test.py：

import asyncio

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

在Azure函数的handler中调用：

import azure.functions as func
from file import test

async def main(req: func.HttpRequest) -> func.HttpResponse:
    # 在函数处理逻辑中执行异步代码
    my_test = await test()
    # 后续业务逻辑
    return func.HttpResponse(f"Result: {my_test}")

这种方式完全适配Azure的异步运行模型，避免了事件循环冲突，也是云函数开发的最佳实践。

方案二：修复自定义运行函数（适配模块级别执行） #

如果必须在模块级别完成异步初始化，可修改自定义my_run函数，正确处理本地和Azure环境的事件循环：

import asyncio
from typing import TypeVar, Callable, Any

T = TypeVar('T')

def my_run(coro_func: Callable[..., T], *args, **kwargs) -> T:
    """适配本地与Azure环境的异步协程运行工具"""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # 本地环境：无运行中循环，直接用asyncio.run创建新循环执行
        return asyncio.run(coro_func(*args, **kwargs))
    else:
        # Azure环境：已有运行中循环，通过线程安全方式提交协程并同步等待结果
        future = asyncio.run_coroutine_threadsafe(coro_func(*args, **kwargs), loop)
        return future.result()

调用方式（注意传协程函数而非已实例化的协程对象）：

# test.py中
my_test = my_run(test)  # 传入test函数，而非test()

方案三：延迟初始化 #

通过惰性加载方式，在第一次使用结果时再执行异步代码，避免模块加载时的冲突：

import asyncio

async def test():
    await asyncio.sleep(1)
    return "test"

_my_test_cache = None

def get_my_test():
    global _my_test_cache
    if _my_test_cache is None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            _my_test_cache = asyncio.run(test())
        else:
            future = asyncio.run_coroutine_threadsafe(test(), loop)
            _my_test_cache = future.result()
    return _my_test_cache

在业务代码中使用：

from file import get_my_test

# 第一次调用时触发异步初始化
my_test = get_my_test()

内容的提问来源于stack exchange，提问作者Ziur Olpa