AWS CDK でStep FunctionsのAPIを作成してみた。

2022.02.21

概要

この記事では、AWS CDKで同期エクスプレスステートマシンを使用して API Gateway REST API を作成しました。ここでは、StepFunctionsを呼び出すためのHTTP「ANY」メソッドAPIを実装しました。StepFunctionsRestApiコンストラクトを使用して、ステートマシンをAPI Gatewayに接続しました。HTTPS リクエストが API メソッドに送信されると、API Gateway はStep Functions API アクションを呼び出します。

 

 

やってみた

CDKアプリの作成

  • 次のコマンドを使用してCDKをインストールしておきます。
python -m pip install aws-cdk-lib

 

  • 新しいディレクトリを作成しておきます。
  • CDKは、プロジェクトディレクトリの名前に基づいてソースファイルとクラスに名前を付けます。
#create new directory
mkdir api-stepfunctions
cd api-stepfunctions

 

  • cdk initコマンドを使用してアプリを初期化しておきます。
cdk init --language python

 

  • 次の2つのコマンドを実行して、アプリのPython virtual environmentをアクティブ化し、CDK core dependenciesをインストールしておきます。
source .venv/bin/activate
pip install -r requirements.txt

 

サービスの作成

  • 新しいファイル [api_stepfunctions/services.py] を作成して、作成する必要のあるAWSサービスを定義しておきます。

 

ステートマシンを定義する

  • パス状態でMachine Definition を定義しておきます。
invoke_state = stepfunctions.Pass(self,"InvokeState", 
              result = stepfunctions.Result("API Gateway Invokes Step Functions"))

success_state = stepfunctions.Pass(self,"SuccessState", 
               result = stepfunctions.Result("Successfully invoked"))

machine_definition = invoke_state.next(success_state)

 

  • Machine Definition とEXPRESSタイプでステートマシンを定義しておきます。
  • StepFunctionsRestApiは同期Expressステートマシンのみを許可するため、ステートマシンタイプはEXPRESSとして設定されます。

 

state_machine = stepfunctions.StateMachine(self, 'StateMachine', 
                definition = machine_definition, 
                state_machine_name = "ApiStepFunctions",
                state_machine_type = stepfunctions.StateMachineType.EXPRESS)

 

REST APIを定義する

  • StepFunctionsRestApiコンストラクトを使用してAPIGateway RESTAPIを作成します。

 

api = apigateway.StepFunctionsRestApi(self, "StepFunctionsApi",
      deploy=True,
      state_machine=state_machine)

 

完全なコード

  • すべてのサービスが定義された完全なコードです。
import aws_cdk as cdk
from constructs import Construct
from aws_cdk import (
    aws_stepfunctions as stepfunctions,
    aws_apigateway as apigateway
)

class serviceStack(Construct):
    def __init__(self, scope: Construct, id: str):
        super().__init__(scope, id)

        invoke_state = stepfunctions.Pass(self,"InvokeState", 
                                result = stepfunctions.Result("API Gateway Invokes Step Functions"))

        success_state = stepfunctions.Pass(self,"SuccessState", 
                                result = stepfunctions.Result("Successfull invoked"))

        machine_definition = invoke_state.next(success_state)

        state_machine = stepfunctions.StateMachine(self, 'StateMachine', 
                definition = machine_definition, 
                state_machine_name = "ApiStepFunctions",
                state_machine_type = stepfunctions.StateMachineType.EXPRESS)

        api = apigateway.StepFunctionsRestApi(self, "StepFunctionsApi",
                deploy=True,
                state_machine=state_machine)

 

アプリにサービスを追加する

  • /api_stepfunctions/api_stepfunctions_stack.py ファイルに次のコードを追加しておきます。
#Import the Service file created in the previou step
from . import services

#service_file.constructor
services.serviceStack(self, "api-invoke-step-functions")

 

CDK Deploy

  • Deploy する前に、環境をブートストラップする必要があります。
  • 次のコマンドを実行して、AWS環境をブートストラップしておきます。
cdk bootstrap

 

  • 次のコマンドを使用してCDKを展開しておきます。
cdk deploy

 

  • コンソールでは、サービスが作成されたことを見ることができます。

State Machine

 

State Machine Definition

 

API Gateway

 

APIをテストする

  • API Gatewayのコンソールで、APIを選択して、Resources ペインで、テストするメソッドを選択しておきます。
  • Method Execution ペインで、Testを選択しておきます。
  • POSTメソッドを選択して、Testをクリックしておきます。
  • Step Functionsの出力を見ることができます。

 

 

まとめ

AWS CDK でStep FunctionsのAPIを作成してみました。

Reference :

Step Function Rest API using CDK

CDK Serverless