如何将Azure Data Factory输入变量传递给Durable Function

1. 先修复现有客户端函数的语法错误 #

你贴的客户端函数代码存在语法问题：参数获取行多写了一个右括号，先修正为可正常运行的版本：

import logging
import azure.functions as func
import azure.durable_functions as df
log = logging.getLogger(__name__)

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    # 原代码此处多了一个)，已修正
    args= {"amount":req.params.get("amount")}

    instance_id = await client.start_new("DurableFunctionsOrchestrator", None, args)
    return client.create_check_status_response(req, instance_id)

2. 匹配现有查询参数读取逻辑的ADF配置 #

你当前代码通过req.params.get()读取URL查询参数，不需要修改函数代码，直接在ADF的Azure Function活动块按以下步骤配置：

• 打开画布上对应调用客户端函数的Azure Function活动配置面板
• 切换到「设置」标签页
• 请求方法选择GET（POST请求也支持携带查询参数，GET更契合查询参数传参的使用习惯）
• 在「查询参数」配置区域新增一条规则：
  • 名称列填amount
  • 值列填ADF表达式 @pipeline().parameters.amount，直接引用你已经提前定义好的ADF管道参数
• 函数路径配置保持你原有设置即可，不需要额外修改。

3. 可选：请求Body传参方案（适合复杂参数场景） #

如果后续需要传递多字段、结构化的参数，不想把参数拼接在URL中，可以修改客户端函数为读取请求Body的模式，再在ADF中配置Body传参：

3.1 修改客户端函数代码 #

import logging
import azure.functions as func
import azure.durable_functions as df
log = logging.getLogger(__name__)

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    # 从请求体读取JSON格式参数
    req_body = req.get_json()
    args= {"amount": req_body.get("amount")}

    instance_id = await client.start_new("DurableFunctionsOrchestrator", None, args)
    return client.create_check_status_response(req, instance_id)

3.2 ADF侧配置 #

• 打开Azure Function活动的「设置」标签页
• 请求方法选择POST
• 在「请求正文」配置项填入对应表达式：
  • 如果amount是字符串类型：

    @json('{"amount":"' + pipeline().parameters.amount + '"}')
    

  • 如果amount是数值类型：

    @json('{"amount":' + string(pipeline().parameters.amount) + '}')
    

• 查询参数区域不需要额外配置。

4. 编排器函数接收参数验证 #

两种传参方式最终传入start_new的args对象，都会作为编排器的入参存在，编排器中可以直接通过上下文获取，示例写法：

def orchestrator_function(context: df.DurableOrchestrationContext):
    input_params = context.get_input()
    amount_val = input_params["amount"]
    # 后续编写具体编排逻辑即可
main = df.Orchestrator.create(orchestrator_function)

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