如何向Azure Durable Activity Function传递多个参数

Question

我的编排程序接收一个有效载荷，有效载荷中包含需要传递到活动函数的指令和其他数据集。

如何将多个参数传递给活动函数？还是必须将所有数据混合在一起？

def orchestrator_function(context: df.DurableOrchestrationContext):
    
    # User defined configuration
    instructions: str = context.get_input()

    task_batch = yield context.call_activity("get_tasks", None)
    
    # Need to pass in instructions too
    parallel_tasks = [context.call_activity("perform_task", task) for task in task_batch]

    results = yield context.task_all(parallel_tasks)
    
    return results

perform_task活动需要来自task_batch和用户输入instructions的两个项目。

我在我的function.json中做了什么吗？

解决方法不是很理想，但是我可以将多个参数作为单个元组传递。

something = yield context.call_activity("activity", ("param_1", "param_2"))

然后，我只需要在活动中引用参数的正确索引。

- Ari

Answer 1

似乎没有一种标准的方法来实现它。我选择给我的单个参数取一个通用名称，例如parameter或payload。

然后，在编排器中传递值时，我是这样做的：

payload = {"value_1": some_var, "value_2": another_var}
something = yield context.call_activity("activity", payload)

然后在activity函数内部，我再次对其进行解包。

编辑：一些埋藏的文档似乎表明 https://learn.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-error-handling?tabs=python

- Ari

---

太棒了！ - Ash

如何在activity函数中解包参数？ - Pingpong

@Pingpong 这只是一个字典，所以在我的例子中，我会这样做 value = payload["value_1"] 并且payload作为参数传递给活动。 - Ari

Answer 2

补充@Ari的很棒的回答，这里有一个示例代码，可将数据从客户端函数（在本例中为HTTP请求）传递到活动函数：

客户端 -> 编排器 -> 活动

客户端

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    req_data = req.get_json()
    img_url = req_data['img_url']
    payload = {"img_url": img_url}
    
    instance_id = await client.start_new(req.route_params["functionName"], None, payload)

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

    return client.create_check_status_response(req, instance_id)

编排器

def orchestrator_function(context: df.DurableOrchestrationContext):
    input_context = context.get_input()
    img_url = input_context.get('img_url')

    some_response= yield context.call_activity('MyActivity', img_url)
    
    return [some_response]

活动

def main(imgUrl: str) -> str:
    print(f'.... Image URL = {imgUrl}')

    return imgUrl

- gary

Answer 3

您可以使用 @dataclass 和 @dataclass_json 类装饰器来定义您的输入和输出类型，例如：

@dataclass_json
@dataclass
class Command:
  param1: str
  param2: int


@dataclass_json
@dataclass
class Result:
  val1: str
  val2: int

然后您可以在 Azure Functions 中使用这些内容，例如在活动函数中：

def main(input: DownloadFileRequest) -> DownloadFileResponse:
  # function code
  result: DownloadFileResponse = DownloadFileResponse("some", 123)
  return result

这为您提供了一个干净的API和描述性代码。对我来说，这比使用字典要好得多。

- Kokos

Answer 4

我建议您考虑使用dataclass-wizard，这是一个可行的选择，相比于dataclasses-json来说更加轻量级；它更轻量级的原因在于不像marshmallow那样使用外部库来生成模式。此外，它的性能也稍微好一些。

很遗憾，我并没有完全理解问题中所概述的数据结构，因此我决定为了快速地展示一个简单的演示而自己编写了一个模式，请查看：

from dataclasses import dataclass
from dataclass_wizard import JSONWizard

@dataclass
class Batman(JSONWizard):
  suit_color: str
  side: 'Sidekick'


@dataclass
class Sidekick:
  name: str
  mood: str

# usage
bats = Batman.from_dict({'suitColor': 'BLACK',
                         'side': {'Name': 'Robin', 'mood': 'Mildly depressed'}})

print(repr(bats)) 
# prints:
#   Batman(suit_color='BLACK', side=Sidekick(name='Robin', mood='Mildly depressed'))
print(bats.to_dict())
# prints:
#   {'suitColor': 'BLACK', 'side': {'name': 'Robin', 'mood': 'Mildly depressed'}}

- rv.kvetch

Answer 5

我从未在Python中使用过Azure函数，只在C#中使用过，但我认为除了将值作为字典传递之外，您还可以将它们作为元组传递，然后在函数内部进行重构。

- Michael Murphy