如何向Python版Durable Azure Functions提交JSON数据并完成日志调试

1. 修复HTTP Starter函数，支持POST请求并传递JSON #

你之前的代码有两个问题：一是只读取了GET请求的URL参数，二是start_new方法参数顺序错误（第二个参数是自定义实例ID，第三个才是传给编排函数的输入数据）。
修改后的代码如下：

import logging
import json

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)
    
    # 兼容GET和POST请求获取输入数据
    try:
        # 优先读取POST请求的JSON body
        input_data = req.get_json()
    except ValueError:
        # 如果是GET请求，读取URL参数中的jsoninput
        jsoninput_str = req.params.get('jsoninput')
        input_data = json.loads(jsoninput_str) if jsoninput_str else {}
    
    # 第三个参数为传给编排函数的输入数据，第二个参数传None自动生成实例ID
    instance_id = await client.start_new(req.route_params["functionName"], None, input_data)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    logging.info(f"传入的输入数据: {json.dumps(input_data, ensure_ascii=False)}")

    return client.create_check_status_response(req, instance_id)

同时要确保Starter函数的function.json配置中允许POST请求，找到对应目录下的function.json，修改methods字段：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "methods": [
        "get",
        "post"
      ],
      "route": "orchestrators/{functionName}"
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

2. 修改编排函数，接收JSON并传递给Activity Function #

编排函数通过context.get_input()获取从Starter传过来的JSON数据，再传递给Activity：

import logging
import json
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    logging.info("进入编排函数")
    # 获取Starter传入的JSON数据
    input_json = context.get_input()
    logging.info(f"编排函数收到的输入: {json.dumps(input_json, ensure_ascii=False)}")
    
    # 将JSON传递给Activity Function，也可以按需传递JSON中的某个字段
    result = yield context.call_activity('CalculateActivity', input_json)
    return result

main = df.Orchestrator.create(orchestrator_function)

如果你需要调用多次Activity，也可以把JSON数组传过来循环调用，按需调整即可。

3. 修改Activity Function，打印接收的JSON数据 #

直接接收传入的字典对象，用logging打印即可：

import logging
import json

def main(input_data: dict) -> str:
    # 打印完整的JSON内容
    logging.info(f"Activity收到的输入数据: {json.dumps(input_data, ensure_ascii=False)}")
    # 也可以打印指定字段，比如input_data.get("name")
    return f"处理完成，收到的名称为: {input_data.get('name', '未传入')}"

测试方法 #

发送POST请求到Starter的URL，请求头设置Content-Type: application/json，请求体填入你要传的JSON即可，例如：

{
  "name": "测试用户",
  "age": 25,
  "location": "北京"
}

查看函数日志即可看到各环节打印的JSON内容。

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