使用pulumi_azure_native.authorization时遇ThreadPoolExecutor线程无事件循环错误

问题 #

我在项目里写了个函数，用来获取当前Pulumi登录上下文的凭证，给自定义提供程序组件使用，代码如下：

from typing import Optional, Tuple
import pulumi_azure_native.authorization as authorization

def get_access_token(endpoint: Optional[str] = None) -> Tuple[str, str, str]:
    access_token = authorization.get_client_token(endpoint=endpoint).token
    client_config = authorization.get_client_config()
    tenant_id = client_config.tenant_id
    subscription_id = client_config.subscription_id
    return tenant_id, subscription_id, access_token

但执行pulumi up时直接触发错误：

C:\Workspace\vsts-agent-win-x64-2.191.1\_work\39\s\Infrastructure\venv\lib\site-packages\grpc\_server.py:457: RuntimeWarning: coroutine 'invoke.<locals>.do_rpc' was never awaited
  return None, False
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

pulumi-python:dynamic/keyvault:AccessPolicy (kv-access-policy-pulumi-developer):
  error: Exception calling application: There is no current event loop in thread 'ThreadPoolExecutor-0_0'.

pulumi-python:dynamic/keyvault:Certificate (kv-certificate-auto-ssl-certificate):
  error: Exception calling application: There is no current event loop in thread 'ThreadPoolExecutor-0_0'.

我原本以为pulumi_azure_native.authorization的方法会按文档描述正常工作，但现在找不到便捷方式在代码中访问当前Pulumi登录上下文。

解决方案 #

这个错误是因为Pulumi动态组件的代码运行在ThreadPoolExecutor的子线程中，子线程默认没有Python事件循环，但authorization.get_client_token和get_client_config底层依赖异步操作，必须有事件循环才能执行。

方法一：手动创建事件循环适配子线程 #

在函数内手动初始化并运行事件循环，完成异步操作后关闭：

from typing import Optional, Tuple
import asyncio
import pulumi_azure_native.authorization as authorization

def get_access_token(endpoint: Optional[str] = None) -> Tuple[str, str, str]:
    # 为子线程创建事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    # 同步执行异步授权方法
    client_token = loop.run_until_complete(authorization.get_client_token(endpoint=endpoint))
    client_config = loop.run_until_complete(authorization.get_client_config())
    
    tenant_id = client_config.tenant_id
    subscription_id = client_config.subscription_id
    access_token = client_token.token
    
    loop.close()
    return tenant_id, subscription_id, access_token

方法二：使用Pulumi的invoke方法处理异步上下文 #

Pulumi的invoke方法会自动处理异步操作的上下文，适配不同线程环境：

from pulumi import invoke
from typing import Optional, Tuple
import pulumi_azure_native.authorization as authorization

def get_access_token(endpoint: Optional[str] = None) -> Tuple[str, str, str]:
    # 用invoke同步调用授权方法，自动处理事件循环
    token_result = invoke(authorization.get_client_token, {"endpoint": endpoint})
    config_result = invoke(authorization.get_client_config)
    
    # 提取实际值（若在动态组件中需注意Output类型的处理逻辑）
    access_token = token_result.token
    client_config = config_result.value
    
    return client_config.tenant_id, client_config.subscription_id, access_token

注意：如果是在常规Pulumi代码（非动态组件）中调用authorization.get_client_token，Pulumi会自动处理事件循环，不会出现这个问题，只有动态组件的子线程环境需要额外适配。

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