如何向Python版Durable Azure Function POST JSON数据并对接ADF

核心代码修改说明 #

你需要依次修改HTTP Starter、编排器、活动函数三类函数，实现参数从ADF到业务逻辑的透传：

1. HTTP Starter函数修改 #

作用是接收ADF POST的JSON数据，透传给编排器，修改点是把start_new方法的第三个参数从空改为请求体内容：

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

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    # 读取ADF POST的JSON参数
    req_payload = req.get_json()
    # 把读取到的参数作为第三个参数传入start_new方法
    instance_id = await client.start_new(req.route_params["functionName"], None, req_payload)
    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)

2. 编排器(Orchestrator)函数修改 #

作用是提取传入的JSON参数，传递给活动函数做业务处理，修改点是通过context.get_input()获取透传的参数，提取需要的字段：

import logging
import json
import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    # 获取HTTP Starter透传的JSON参数
    input_params = context.get_input()
    # 提取参数中的Country字段，示例值为"Japan"
    target_country = input_params.get("Country", "")
    # 把Country参数传递给业务活动函数，可根据需求扩展多活动调用
    process_result = yield context.call_activity('Hello', target_country)
    # 最终返回处理结果供ADF读取
    return process_result

main = df.Orchestrator.create(orchestrator_function)

3. 活动(Activity)函数修改 #

作用是接收编排器传递的参数，执行具体的业务逻辑（比如按Country过滤结果集），示例修改如下：

import logging

def main(target_country: str) -> dict:
    # 此处替换为你的实际业务逻辑，比如按传入的Country过滤结果集
    logging.info(f"正在过滤国家为 {target_country} 的数据")
    # 示例返回处理结果，ADF可直接读取该返回值
    return {
        "status": "success",
        "filtered_count": 100,
        "target_country": target_country
    }

ADF侧配置（保证10分钟任务无超时） #

• 新增Azure Function活动，请求方法选择POST，请求Body填写你需要传递的JSON参数，示例：{"Country": "Japan"}
• 活动超时时间设置为大于任务预期运行时长，比如设置为00:15:00（15分钟）
• 无需额外配置轮询逻辑：Durable Functions返回的202状态码包含状态查询地址，ADF的Function活动会自动识别并定期轮询任务状态，直到运行完成或失败，不会出现HTTP连接超时问题。

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