如何在Azure Durable Python Functions中使用托管标识实现队列触发器

步骤1：配置托管身份与存储权限 #

• 启用函数应用的系统分配托管身份：在Azure门户的函数应用→身份→系统分配→状态设为“开启”并保存。
• 给托管身份分配队列权限：进入目标存储账户→访问控制(IAM)→添加角色分配，选择存储队列数据参与者角色，成员类型选“托管身份”，然后选择你的函数应用身份，完成分配。

步骤2：修改函数配置 #

• 移除原有的QueueConnectionString环境变量，或者添加新的应用设置：
  • 若直接使用存储账户名，无需额外设置；
  • 若需指定队列服务URI，添加格式为<AccountAlias>__queueServiceUri的应用设置，值为https://<你的存储账户名>.queue.core.windows.net/（后续connection参数填<AccountAlias>）。

步骤3：调整函数代码 #

将队列触发器装饰器的connection参数从连接字符串配置名改为存储账户名或你设置的<AccountAlias>：

import azure.functions as func
import logging
import azure.durable_functions as adf

myApp = adf.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@myApp.durable_client_input(client_name="client")
# 替换connection值为你的存储账户名或配置别名
@myApp.queue_trigger(arg_name="azqueue", queue_name="test", connection="你的存储账户名")
async def begin_data_entry(azqueue: func.QueueMessage, client):
    logging.info('Python队列触发器处理了一条请求。')
    await client.start_new("activity_function_name", client_input={})

步骤4：配置Durable Functions的托管身份（可选） #

若Durable Functions的后台存储也想改用托管身份，修改host.json：

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "storageProvider": {
        "type": "azureStorage",
        "connectionStringName": "你的存储账户名",
        "useManagedIdentity": true
      }
    }
  }
}

本地开发注意事项 #

本地调试时，确保你有通过Azure CLI、Visual Studio或其他方式登录Azure，DefaultAzureCredential会自动选取合适的身份进行认证。如果需要，可安装依赖包：

pip install azure-identity

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