python - 如何将 JSON 发布到持久 Azure Function (Python)？

Question

我想从 Azure 数据工厂调用持久 Azure Functions。 我想将 json 发布到 Functions 并在函数处理完成时获取状态。 我的最终目标是成功运行需要 10 分钟且不超时的函数。

我已经使用方法 GET 从 ADF 成功执行了 Azure Function Activity。

现在我需要建议修改 Orchestrator 的 Python 代码以接受 json 并使用 json 值来过滤处理哪个结果集。 {“国家”:“日本”}

当前代码库来自教程: https://learn.microsoft.com/en-us/azure/azure-functions/durable/quickstart-python-vscode

我正在遵循此处的耐用功能说明: http://datanrg.blogspot.com/2020/10/using-durable-functions-in-azure-data.html

# This function an HTTP starter function for Durable Functions.
# Before running this sample, please:
# - create a Durable orchestration function
# - create a Durable activity function (default name is "Hello")
# - add azure-functions-durable to requirements.txt
# - run pip install -r requirements.txt

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)
instance_id = await client.start_new(req.route_params["functionName"], None, None)

logging.info(f"Started orchestration with ID = '{instance_id}'.")

return client.create_check_status_response(req, instance_id)

#  This function is not intended to be invoked directly. Instead it will be
# triggered by an HTTP starter function.
# Before running this sample, please:
# - create a Durable activity function (default name is "Hello")
# - create a Durable HTTP starter function
# - add azure-functions-durable to requirements.txt
# - run pip install -r requirements.txt

import logging
import json

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


def orchestrator_function(context: df.DurableOrchestrationContext):
result1 = yield context.call_activity('Hello', "Tokyo")
result2 = yield context.call_activity('Hello', "Seattle")
result3 = yield context.call_activity('Hello', "London")
return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)


# This function is not intended to be invoked directly. Instead it will be
# triggered by an orchestrator function.
# Before running this sample, please:
# - create a Durable orchestration function
# - create a Durable HTTP starter function
# - add azure-functions-durable to requirements.txt
# - run pip install -r requirements.txt

import logging


def main(name: str) -> str:
    return f"Hello {name}!"

Answer 1

Now I need advice to modify Python code of Orchestrator to accept Json and use Json values to filter which Result set is processed. {"Country": "Japan"}

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


async def main(documents: func.DocumentList, starter: str):
    client = df.DurableOrchestrationClient(starter)
    instance_id = await client.start_new('MyDFOrchestrator', {"doc_list": [{doc1}, {doc2}, {doc3}]})
    logging.info(f"Started orchestration ID {instance_id}")

JSON 应该可以作为输入值提供给协调器。这是一个example实现了类似的目标。尽管该示例使用了 http 触发器，但必须寻址的区域与您选择在启动器或触发函数中使用的触发器无关。

作为替代方案，您可以开发一个可序列化并具有模型/实体结构的具体类(比原始 Json 干净得多)。我们所需要的只是让你的类导出两个静态方法，到 Json() 和从 Json 导出，以生成可序列化的类 ()。这些类将由 Durable Functions 框架重复调用，以序列化和反序列化您的自定义类。