AWS Lambda Durable Functionで並列操作parallelをためしてみた #AWSreInvent

def parallel(
    functions: Sequence[Callable[[DurableContext], T]],
    name: str | None = None,
    config: ParallelConfig | None = None,
) -> BatchResult[T]

result: BatchResult[dict] = context.parallel([task1, task2, task3])

# 成功した結果を取得（入力順序を保持）
first_result = result.successful_results[0]
second_result = result.successful_results[1]
third_result = result.successful_results[2]

# 失敗があるかチェック
if result.failure_count > 0:
    print(f"失敗したBranch数: {result.failure_count}")
    print(f"失敗結果: {result.failed_results}")

# カウントの確認
print(f"総Branch数: {result.total_count}")
print(f"成功数: {result.success_count}")

API アクションの呼び出しに失敗しました。エラーメッセージ: { "errorMessage": "'BatchResult' object has no attribute 'allsuccessful_results'", "errorType": "AttributeError" }

from aws_durable_execution_sdk_python.config import ParallelConfig

config = ParallelConfig(max_concurrency=5)  # 最大5つまで同時実行
result = context.parallel(functions, config=config)

from aws_durable_execution_sdk_python.config import CompletionConfig, ParallelConfig

# 最初に成功したものだけ取得（複数のデータソースから取得する場合など）
config = ParallelConfig(
    completion_config=CompletionConfig.first_successful()
)

# すべて完了まで待機（一部失敗しても結果を収集したい場合）
config = ParallelConfig(
    completion_config=CompletionConfig.all_completed()
)

import time

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)

@durable_execution
def lambda_handler(event: dict, context: DurableContext) -> dict:
    """3つのAPIを並列で呼び出して結果を統合する"""

    # 各APIを呼び出す関数を定義
    def fetch_user(ctx: DurableContext) -> dict:
        # ユーザー情報APIの応答時間（2秒）をシミュレート
        def call_user_api(_):
            time.sleep(2)  
            return {"user_id": "U001", "name": "Taro Yamada", "email": "taro@example.com"}
        return ctx.step(call_user_api, name="fetch_user")

    def fetch_orders(ctx: DurableContext) -> dict:
        # 注文情報APIの応答時間（3秒）をシミュレート
        def call_orders_api(_):
            time.sleep(3)  
            return {"order_id": "O001", "items": ["item1", "item2"], "total": 5000}
        return ctx.step(call_orders_api, name="fetch_orders")

    def fetch_inventory(ctx: DurableContext) -> dict:
        # 在庫情報APIの応答時間（2秒）をシミュレート
        def call_inventory_api(_):
            time.sleep(2)  
            return {"product_id": "P001", "stock": 100, "warehouse": "Tokyo"}
        return ctx.step(call_inventory_api, name="fetch_inventory")

    # 3つのAPIを並列で実行

    result: BatchResult[dict] = context.parallel([
        fetch_user,
        fetch_orders,
        fetch_inventory,
    ])

    # 結果を統合して返却

    return {
        # result
        # "user": result.allsuccessful_results[0],
        # "orders": result.successful_results[1],
        # "inventory": result.successful_results[2],

        "user": result.all[0].result,
        "orders": result.all[1].result,
        "inventory": result.all[2].result,

        # Counts
        "total_count": result.total_count,
        "success_count": result.success_count,
        "failure_count": result.failure_count,

        # Status information
        "status": result.status.value,
        "completion_reason": result.completion_reason.value,
    }

    def fetch_inventory(ctx: DurableContext) -> dict:
        # 在庫情報APIの応答時間（2秒）をシミュレート
        def call_inventory_api(_):
            time.sleep(2)  
            return {"product_id": "P001", "stock": 100, "warehouse": "Tokyo"}
        raise Exception("エラーです。")
        return ctx.step(call_inventory_api, name="fetch_inventory")

    return {
    ...
        "inventory": str(result.all[2].error),
    ...

{
  "user": {
    "user_id": "U001",
    "name": "Taro Yamada",
    "email": "taro@example.com"
  },
  "orders": {
    "order_id": "O001",
    "items": [
      "item1",
      "item2"
    ],
    "total": 5000
  },
  "inventory": "ErrorObject(message='エラー', type='CallableRuntimeError', data=None, stack_trace=None)",
  "total_count": 3,
  "success_count": 2,
  "failure_count": 1,
  "status": "FAILED",
  "completion_reason": "ALL_COMPLETED"
}