如何通过Durable Functions将数据输出至Azure Blob服务

问题根源 #

原代码的核心错误在于Durable Functions活动函数无法直接使用@app.blob_output输出绑定。活动函数由Durable Task Framework调度，这类绑定会与Durable的执行模型冲突，导致运行时错误。正确的做法是在活动函数内通过Azure Blob Storage SDK手动完成Blob写入操作。

修改后完整代码 #

import azure.functions as func
import azure.durable_functions as df
import logging
from azure.storage.blob import BlobServiceClient
import os

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

### 客户端函数 ###
@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async def client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    instance_id = await client.start_new("orchestrator", None, {})
    logging.info(f"启动编排实例ID: '{instance_id}'.")
    # 等待编排完成并返回标准响应
    return await client.wait_for_completion_or_create_check_status_response(req, instance_id)

### 编排器函数 ###
@app.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext) -> dict:
    # 调用活动函数，可传入计算参数（示例传空）
    activity_result = yield context.call_activity("calculate_and_save_to_blob", "")
    return {"执行结果": activity_result}

### 活动函数（计算+写入Blob） ###
@app.activity_trigger(input_name="input_params")
def calculate_and_save_to_blob(input_params: str) -> str:
    # 1. 模拟数值计算逻辑，替换为实际业务代码
    calculated_data = "Data is successfully Inserted"
    logging.info(f"生成计算数据: {calculated_data}")

    # 2. 连接Blob存储并写入数据
    try:
        # 从环境变量读取Blob存储连接字符串
        conn_str = os.getenv("BlobStorageConnection")
        blob_service_client = BlobServiceClient.from_connection_string(conn_str)
        
        # 获取容器客户端，容器不存在则自动创建
        container_name = "newblob"
        container_client = blob_service_client.get_container_client(container_name)
        container_client.create_container(exists_ok=True)
        
        # 写入Blob内容，覆盖已有文件
        blob_name = "test.txt"
        blob_client = container_client.get_blob_client(blob_name)
        blob_client.upload_blob(calculated_data, overwrite=True)
        
        logging.info(f"成功写入Blob: {container_name}/{blob_name}")
        return "完成：数据已保存至Blob存储"
    except Exception as e:
        logging.error(f"写入Blob失败: {str(e)}")
        raise e

local.settings.json 配置修正 #

确保BlobStorageConnection为完整的存储账户连接字符串，补充截断的内容：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "BlobStorageConnection": "DefaultEndpointsProtocol=https;AccountName=你的存储账户名;AccountKey=你的存储账户密钥;EndpointSuffix=core.windows.net"
  }
}

关键修改点 #

1. 移除活动函数上的@app.blob_output绑定，改用azure-storage-blob SDK实现Blob操作
2. 在活动函数内添加Blob存储连接、容器自动创建、数据写入的完整逻辑
3. 优化客户端函数返回逻辑，使用Durable内置方法返回标准编排响应
4. 增加异常捕获与日志记录，便于问题排查

依赖安装 #

执行以下命令安装Blob存储SDK：

pip install azure-storage-blob

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