はじめに
みずほリサーチ＆テクノロジーズ株式会社 先端技術研究部の@yosuke_kosakaです。
今回は、AWS ChaliceとAWS CDKの統合の発展形として、AWS CDKのStep FunctionsのL2 Constructを利用して、Chaliceで定義したAPIとStep Functionsを統合しました。
そもそもAWS Chaliceとは？
サーバーレスアプリケーションをAWSで作成するためのPythonのフレームワークです。
少量のコードで、API GatewayとLambdaを数分で作成してくれます。
【参考】
Chalice公式
爆速 API 開発を実現するサーバーレスアプリケーション開発向けフレームワーク
前提
ChaliceのQuick Startの序盤の設定が完了しているものとします。
AWS ChaliceとAWS CDKの統合
Chalice単独で利用する場合、chalice deployのコマンドでAPIのデプロイが可能ですが、
CDKを利用することで、DynamoDB等、他の資産も含めてデプロイすることが可能となります。
【参考】
Deploying with the AWS CDK — AWS Chalice
CDKを利用したChaliceプロジェクトの作成
Chaliceを、AWS CDKを利用してデプロイします。
Chaliceプロジェクトの新規作成時、[CDK] Rest API with a DynamoDB tableを選ぶことでCDKと統合されたプロジェクトが作成されます。
$ chalice new-project
___ _ _ _ _ ___ ___ ___
/ __|| || | /_\ | | |_ _|/ __|| __|
| (__ | __ | / _ \ | |__ | || (__ | _|
\___||_||_|/_/ \_\|____||___|\___||___|
The python serverless microframework for AWS allows
you to quickly create and deploy applications using
Amazon API Gateway and AWS Lambda.
Please enter the project name
[?] Enter the project name: cdk-sfntest
[?] Select your project type: [CDK] Rest API with a DynamoDB table
REST API
S3 Event Handler
Lambda Functions only
Legacy REST API Template
> [CDK] Rest API with a DynamoDB table
Your project has been generated in ./cdk-sfntest
フォルダ構成
$ tree
.
├── infrastructure
│ ├── app.py
│ ├── cdk.json
│ ├── requirements.txt
│ └── stacks
│ ├── chaliceapp.py
│ └── __init__.py
├── README.rst
├── requirements.txt
└── runtime
├── app.py
└── requirements.txt
3 directories, 9 files
runtime/app.py(Chaliceのコード)には以下のようなコードが初期設定されています。
- create_userメソッド
- get_userメソッド
import os
import boto3
from chalice import Chalice
app = Chalice(app_name='cdk-sfntest')
dynamodb = boto3.resource('dynamodb')
dynamodb_table = dynamodb.Table(os.environ.get('APP_TABLE_NAME', ''))
@app.route('/users', methods=['POST'])
def create_user():
request = app.current_request.json_body
item = {
'PK': 'User#%s' % request['username'],
'SK': 'Profile#%s' % request['username'],
}
item.update(request)
dynamodb_table.put_item(Item=item)
return {}
@app.route('/users/{username}', methods=['GET'])
def get_user(username):
key = {
'PK': 'User#%s' % username,
'SK': 'Profile#%s' % username,
}
item = dynamodb_table.get_item(Key=key)['Item']
del item['PK']
del item['SK']
return item
infrastructure/stacks/chaliceapp.pyには以下のコードが初期設定されています。
__init__内でChaliceが定義されており、runtimeフォルダがソースディレクトリとして利用されています。
DynamoDBもこちらで定義されています。
import os
from aws_cdk import aws_dynamodb as dynamodb
try:
from aws_cdk import core as cdk
except ImportError:
import aws_cdk as cdk
from chalice.cdk import Chalice
RUNTIME_SOURCE_DIR = os.path.join(
os.path.dirname(os.path.dirname(__file__)), os.pardir, 'runtime')
class ChaliceApp(cdk.Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
self.dynamodb_table = self._create_ddb_table()
self.chalice = Chalice(
self, 'ChaliceApp', source_dir=RUNTIME_SOURCE_DIR,
stage_config={
'environment_variables': {
'APP_TABLE_NAME': self.dynamodb_table.table_name
}
}
)
self.dynamodb_table.grant_read_write_data(
self.chalice.get_role('DefaultRole')
)
def _create_ddb_table(self):
dynamodb_table = dynamodb.Table(
self, 'AppTable',
partition_key=dynamodb.Attribute(
name='PK', type=dynamodb.AttributeType.STRING),
sort_key=dynamodb.Attribute(
name='SK', type=dynamodb.AttributeType.STRING
),
removal_policy=cdk.RemovalPolicy.DESTROY)
cdk.CfnOutput(self, 'AppTableName',
value=dynamodb_table.table_name)
return dynamodb_table
infrastructure/app.py(CDKのapp.py)には以下のコードが初期設定されています。
CDKでChaliceを利用する場合、stacks.chaliceappからChaliceAppのスタックをインポートして利用することで統合が可能です。
#!/usr/bin/env python3
try:
from aws_cdk import core as cdk
except ImportError:
import aws_cdk as cdk
from stacks.chaliceapp import ChaliceApp
app = cdk.App()
ChaliceApp(app, 'cdk-sfntest')
app.synth()
CDKを利用してデプロイ
必要なライブラリをインストールします。
pip3 install aws-cdk-lib
pip3 install boto3
初期状態のままデプロイしてみます。infrastructureフォルダでcdk deployを実行します。
cd cdk-sfntest/infrastructure
cdk deploy
Do you wish to deploy these changes (y/n)? y
cdk-sfntest: deploying... [1/1]
・・・
✨ Total time: 114.9s
完了すると以下のリソースがCDKによって作成されます。
- API Gateway
- Lambda
- DynamoDB
ChaliceのAPIを、CDKのStep Funtionsで利用する
infrastructure/stacks/sfn_stack.pyを作成し、以下のコードを入力します。
from aws_cdk import (
Stack,
aws_stepfunctions_tasks as tasks,
aws_stepfunctions as sfn,
aws_apigateway,
)
class StepfunctionsStack(Stack):
def __init__(self, scope, id: str, api_stack: any, **kwargs) -> None:
super().__init__(scope, id, **kwargs)
# スタック内のAPIを取得
user_api = aws_apigateway.RestApi.from_rest_api_id(
self, "user", api_stack.chalice.get_resource("RestAPI").ref
)
# ユーザー作成
create_user_task = tasks.CallApiGatewayRestApiEndpoint(
self,
id="ユーザー作成",
api=user_api,
stage_name="api",
method=tasks.HttpMethod.POST,
api_path="/users",
auth_type=tasks.AuthType.IAM_ROLE,
request_body=sfn.TaskInput.from_json_path_at("$"),
input_path="$",
result_path="$.CreateResult",
headers=sfn.TaskInput.from_object({
"Content-Type": [
"application/json"
]
})
)
# ユーザー取得
get_user_task = tasks.CallApiGatewayRestApiEndpoint(
self,
id="ユーザー取得",
api=user_api,
stage_name="api",
method=tasks.HttpMethod.GET,
api_path=sfn.JsonPath.format(
"/users/{}", sfn.JsonPath.string_at("$.username")
),
auth_type=tasks.AuthType.IAM_ROLE,
request_body=sfn.TaskInput.from_json_path_at("$"),
input_path="$",
headers=sfn.TaskInput.from_object({
"Content-Type": [
"application/json"
]
}
)
)
# definition作成
definition = sfn.Chain.start(create_user_task).next(
get_user_task
)
# StateMachine定義
sfn.StateMachine(
self,
"UserCreateStateMachine",
definition=definition,
state_machine_type=sfn.StateMachineType.STANDARD,
)
AWS CDKのStep FunctionsとChalice APIとの統合のポイントは以下です。
get_resource
Chalice #get_resource
によると、get_resource(resource_name)によってChaliceで作成されたリソースをCDKのリソースとして取り込み可能です。
しかし、単純にget_resourceでAPIを取得すると、aws_cdk.cdk.CfnResource型で返されてしまい、L1のコンストラクトになってしまいます。
このままでは、StepFunctionsのL2コンストラクトが使えず、Step Functionsのステートマシン定義はJSONで記入する必要が出てしまいます。
from_rest_api_idの利用
aws_apigateway.RestApiのfrom_rest_api_idメソッドを使うことで、APIのIDから取り込むことが可能です。
これを利用して、以下のように定義しました。
※RESTAPIというリソース名で定義されています。
# スタック内のAPIを取得
user_api = aws_apigateway.RestApi.from_rest_api_id(
self, "user", api_stack.chalice.get_resource("RestAPI").ref
)
taskで利用
tasksの、CallApiGatewayRestApiEndpoint内のapiに、上記from_rest_api_idで取得したuser_apiを指定します。
こうすることでAWS CDKのStep FunctionsとChalice APIとの統合が完了します。
# ユーザー作成
create_user_task = tasks.CallApiGatewayRestApiEndpoint(
self,
id="ユーザー作成",
api=user_api,
stage_name="api",
method=tasks.HttpMethod.POST,
api_path="/users",
auth_type=tasks.AuthType.IAM_ROLE,
request_body=sfn.TaskInput.from_json_path_at("$"),
input_path="$",
result_path="$.CreateResult",
headers=sfn.TaskInput.from_object({
"Content-Type": [
"application/json"
]
})
)
StateMachineを定義
必要なtaskを定義できたら、後はdefinitionとStateMachineを定義して完成です。
JSONで定義するよりも簡単にStateMachineを利用できるので、CDKの恩恵を受けられます。
# definition作成
definition = sfn.Chain.start(create_user_task).next(
get_user_task
)
# StateMachine定義
sfn.StateMachine(
self,
"UserCreateStateMachine",
definition=definition,
state_machine_type=sfn.StateMachineType.STANDARD,
)
infrastructure/app.pyの修正
Chaliceスタックを変数に格納し、作成したStepFunctionsのスタックを追加します。
#!/usr/bin/env python3
try:
from aws_cdk import core as cdk
except ImportError:
import aws_cdk as cdk
from stacks.chaliceapp import ChaliceApp
from stacks.sfn_stack import StepfunctionsStack
app = cdk.App()
# Chaliceのスタックを変数に格納
api_stack = ChaliceApp(app, 'cdk-sfntest')
# ChaliceスタックをStepFunctionsスタックに渡す
StepfunctionsStack(app,'sfn-stack',api_stack)
app.synth()
デプロイ前
pip install -r requirements.txt
cd ./infrastructure
pip install -r requirements.txt
デプロイ
infrastructureフォルダでcdk deployを実行します。
cd infrastructure
cdk deploy --all
✨ Deployment time: 93.08s
Stack ARN:
arn:aws:cloudformation:ap-northeast-1:XXXXXXXXXXXX:stack/sfn-stack/XXXXXXXXXXXXXXXXXXXXXXXXXX
✨ Total time: 106.1s
確認
ステートマシンが作成されていることが確認できます。
※今回はL2コンストラクトを利用してステートマシンを定義することに主観を置いており、作ったステートマシンは単純にAPIを繋いだだけのものですので、動作に関しては検証しておりません。
まとめ
今回はChalice+CDKを利用してステートマシンを作る方法をご紹介しました。
CDKのL2以上のコンストラクタは非常に便利に作られており、利用することで開発効率も上げることが可能ですので、
Step Functionsに限らず、他のサービスとも統合してより効率的に開発していきましょう。