为何无法多次调用asyncio.run()？如何异步运行Graph API相关函数

我正在编写脚本，使用Microsoft Graph API连接并查询Azure Entra ID中的用户数据。需要两个异步函数：一个获取指定组的用户，另一个根据用户ID获取其显示名称。单独用asyncio.run()运行任意一个都正常，但同时运行两个时第二个函数报错，搞不清第二次调用asyncio.run()失败的原因。

主脚本中调用第一个函数的代码：

async def get_all_group_members():
    memberListAll = []
    for group in groupList:
        # print(f"We are trying to run group {group}")
        results = await gq.get_group_user(group['GroupID'], graphServiceClient)
        memberDict = {"GroupName":group["GroupName"], "GroupID":group["GroupID"], "MemberList":results}
        memberListAll.append(memberDict)
    return memberListAll
results = asyncio.run(get_all_group_members())

get_group_user的定义：

memberList = []
members = await graphServiceClient.groups.by_group_id(groupID).members.get()
if members and members.value:
    for member in members.value:
        user = await graphServiceClient.users.by_user_id(member.id).get()
        if user:
            memberList.append(user.display_name)
return memberList

主脚本中调用第二个函数的代码：

async def get_user_name():
    results = await gq.get_user_by_id(member=member,graphServiceClient=graphServiceClient)
    return results

results2 = asyncio.run(get_user_name())

get_user_by_id的定义：

user = await graphServiceClient.users.by_user_id(member).get()
print(user)

单独运行任意一个函数都正常，同时运行第二个就失败，求问原因和修改方案。

为什么多次调用asyncio.run()会失败？ #

asyncio.run()的设计是创建并管理一个全新的事件循环，运行传入的协程直到完成后自动关闭事件循环。第一次调用时，它初始化循环、执行协程，之后关闭整个循环；第二次调用时，要么单线程环境下默认策略无法重建有效循环，要么你的graphServiceClient已绑定到第一次的循环资源上，循环关闭后客户端连接失效，导致报错。另外，asyncio.run()本身是作为异步程序的入口点设计的，不推荐多次调用。

修改方案 #

方案1：合并协程到同一事件循环（推荐） #

创建一个主协程，将两个任务串行或并行执行，仅调用一次asyncio.run()启动主协程。

串行执行（按顺序运行）

async def main():
    # 先执行组成员获取逻辑
    results = await get_all_group_members()
    # 再执行用户名称获取逻辑
    results2 = await get_user_name()
    return results, results2

# 仅调用一次asyncio.run()
final_results, final_results2 = asyncio.run(main())

并行执行（无依赖时提升效率）

如果两个任务没有先后依赖，用asyncio.gather()让它们同时运行：

async def main():
    # 同时启动两个协程
    results, results2 = await asyncio.gather(
        get_all_group_members(),
        get_user_name()
    )
    return results, results2

final_results, final_results2 = asyncio.run(main())

方案2：手动管理事件循环（不推荐） #

若需分开调用，可手动创建并复用事件循环，但这种方式不够简洁，且仍可能存在客户端资源绑定问题：

# 创建并设置事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    # 运行第一个协程
    results = loop.run_until_complete(get_all_group_members())
    # 运行第二个协程
    results2 = loop.run_until_complete(get_user_name())
finally:
    # 最后关闭循环
    loop.close()

额外优化：并行获取用户信息 #

原get_group_user中串行获取用户信息效率较低，可改为并行：

async def get_group_user(groupID, graphServiceClient):
    memberList = []
    members = await graphServiceClient.groups.by_group_id(groupID).members.get()
    if members and members.value:
        # 批量创建获取用户的任务
        tasks = [
            graphServiceClient.users.by_user_id(member.id).get()
            for member in members.value
        ]
        # 并行执行所有任务
        users = await asyncio.gather(*tasks)
        # 过滤有效用户并收集显示名称
        memberList = [user.display_name for user in users if user]
    return memberList

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