Serverless FrameworkでAPIGateway・Lambda・DynamoDBを構築する

Serverless Framework

Serverless FrameworkはAWS LambdaやAmazon APIGatewayなどを利用したサーバーレスなアプリケーションを構築や管理・デプロイするためのツールです。
今回はServerless Frameworkを用いてAPIGateway・Lambda・DynamoDBを構築し管理・デプロイします。
上記の3つのサービスを構築できれば、簡単なRESTAPIを作成することができます。
Serverless Frameworkを用いたLambdaの構築・デプロイは以下の記事で紹介しています。
今回はServerless Frameworkを用いたLambdaの構築とデプロイが完了している前提で進めていきます。

Serverless FrameworkでAWS Lambdaをデプロイ

APIGatewayの構築

Serverless FrameworkではLambda関数のトリガーイベントとして、APIGatewayを作成することが出来ます。
公式ドキュメントに詳細な設定方法が載っています。

serverless.yml

# Lambdaを構築
functions:
  # 関数名を指定
  hello:
    # handler関数を指定
    handler: handler.hello
    timeout: 200
    # Lambdaのイベントトリガーを設定
    events:
      # トリガーとしてAPIGatewayを構築
      - http:
          # リソースを指定
          path: sample/test
          # メソッドを指定
          method: get

eventsキーを追加します。httpキーの値にpathキーでHTTP APIのリソースを指定し、methodキーでリソースのメソッドを指定します。
再度デプロイすると、APIが作成されコンソール画面から確認することができ、作成されたAPIのエンドポイントにアクセスできます。

$ serverless deploy -v
Service Information
service: slf-sample
stage: dev
region: ap-northeast-1
stack: slf-sample-dev
api keys:
  None
endpoints:
  GET - https://XXXXXXXXX.execute-api.ap-northeast-1.amazonaws.com/dev/sample/test
functions:
  hello: slf-sample-dev-hello

$ curl -X GET https://XXXXXXXXX.execute-api.ap-northeast-1.amazonaws.com/dev/sample/test

カスタムリクエストテンプレートの作成

APIGatewayのマッピングテンプレートを利用することで、バックエンドのLambdaでプロキシする情報をカスタマイズすることができます。必要な情報を転送することができるため、バックエンドのLambdaの処理をシンプルにすることができます。

serverless.yml

# Lambdaを構築
functions:
  # 関数名を指定
  hello:
    # handler関数を指定
    handler: handler.hello
    timeout: 200
    # Lambdaのイベントトリガーを設定
    events:
      # トリガーとしてAPIGatewayを構築
      - http:
          # リソースを指定
          path: sample/test
          # メソッドを指定
          method: post
          # 統合サービスをLambdaにする
          integration: lambda
          # カスタムリクエストの作成
          request:
            # カスタムテンプレートの作成
            template:
              application/octet-stream:
                '{"headers":{
                  #foreach($key in $input.params().header.keySet())
                  "$key": "$input.params().header.get($key)"#if($foreach.hasNext),#end
                  #end
                  },
                  "body": "$util.base64Encode($input.json(''$''))"
                  }'

integrationキーの値にlambdaを指定してLambdaでプロキシする設定にします。requestキーにtemplateキーを追加し、マッピングテンプレートを追加します。今回はシンプルに、header情報とbody情報を転送することにします。さらに、body情報をbase64で暗号化します。
Lambdaのhandler.pyをログが出力できるように、以下のように編集します。

handler.py

# coding: utf-8
import json
import logging

# ログの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)


def hello(event, context):
    """
    AWS Lambda Handler関数
    @Param event イベントデータ　APIGatewayからのデータ
    @Param content Lambdaのランタイムデータ
    @return APIGatewayのレスポンスデータ
    """
    # イベントデータの表示
    logger.info('headers:' + str(event['headers']))
    logger.info('body:' + str(event['body']))
    # レスポンスbodyを作成
    body = {
        "message": "Go Serverless v1.0! Your function executed successfully!",
        "input": event
    }
    # レスポンスデータの作成
    response = {
        "statusCode": 200,
        "body": json.dumps(body)
    }

    return response

デプロイ後に、作成されたAPIのエンドポイントにPOSTリクエストを送信し、Lambdaのログを確認すると以下のようなログが出力されています。今回はクライアントツールとしてAdvanced REST clientを使用しています。

送信した情報
header
  Host:XXXXXXXXX.execute-api.ap-northeast-1.amazonaws.com
  Accept-Charset:utf-8
  Authorization:testtest
  Content-Type:application/octet-stream
body
  {"test":"test"}

Lambdaのログ
[INFO]  2018-08-16T02:14:54.637Z    2a01f728-a0fa-11e8-9773-6b735628d930
headers: {'Accept-Charset': 'utf-8', 'Authorization': 'testtest', 'CloudFront-Forwarded-Proto': 'https',
 'CloudFront-Is-Desktop-Viewer': 'true', 'CloudFront-Is-Mobile-Viewer': 'false', 'CloudFront-Is-SmartTV-Viewer': 'false',
 'CloudFront-Is-Tablet-Viewer': 'false', 'CloudFront-Viewer-Country': 'JP', 'content-type': 'application/octet-stream',
 'Host': 'XXXXXXXXX.execute-api.ap-northeast-1.amazonaws.com', 'Via': '1.1 86232f3e7171d72cbd25064a993639d3.cloudfront.net (CloudFront)',
 'X-Amz-Cf-Id': 'kRC9iQhLDfE-he1N_-ke_F3N8ieJcKuvTrwNTF-PxBl2_GjCiNtzDA==', 'X-Amzn-Trace-Id': 'Root=1-5b74de1e-454edc259d00c093259d619a',
 'X-Forwarded-For': '122.215.120.62, 204.246.186.90', 'X-Forwarded-Port': '443', 'X-Forwarded-Proto': 'https'}

[INFO]  2018-08-16T02:14:54.637Z    2a01f728-a0fa-11e8-9773-6b735628d930
body: eyJ0ZXN0IjoidGVzdCJ9

APIGatewayのマッピングテンプレートの情報はLambdaのhandler関数の引数eventに渡ります。
Lambdaのログに送信したheader情報が表示されており、body情報はbase64で暗号化されて表示されていることがわかると思います。

カスタムレスポンステンプレート

単純にLambdaからのレスポンスをAPIGatewayが受け取ってレスポンスを返すと、ステータスコードは200しか返しません。200以外の値を返したい場合は、カスタムレスポンステンプレートでステータスコードを指定する必要があります。
今回は認証エラーを示すステータスコード401を返すことにします。まずは、handler関数を編集します。

handler.py

# coding: utf-8
import json
import logging

# ログの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)


def hello(event, context):
    """
    AWS Lambda Handler関数
    @Param event イベントデータ　APIGatewayからのデータ
    @Param content Lambdaのランタイムデータ
    @return APIGatewayのレスポンスデータ
    """
    # イベントデータの表示
    logger.info('headers:' + str(event['headers']))
    logger.info('body:' + str(event['body']))
    # 認証情報の取得
    authoriztion = str(event['headers']['Authorization'])
    # 独自認証。失敗した場合はExceptionを発生させ、カスタムレスポンスコード401を返す。
    if authoriztion != 'test':
        raise UnAuthorizationError(401,"errorMessage")
    # レスポンスbodyを作成
    body = {
        "message": "Go Serverless v1.0! Your function executed successfully!",
        "input": event
    }
    # レスポンスデータの作成
    response = {
        "statusCode": 200,
        "body": json.dumps(body)
    }

    return response


class UnAuthorizationError(Exception):
   """
   認証失敗の独自Exceptionクラス
   @extends Exceptionクラスを継承
   """
   def __init__(self, code, messages):
       """
       コンストラクタ
       @Param code レスポンスコード
       @Param data レスポンスデータ
       """
       self.code = code
       self.messages = messages

   def __str__(self):
       """
       文字列変換メソッド
       """
       response = {
           'status': 'HTTP/1.1 401 Unauthorized',
           'statusCode': self.code,
           'headers': {
               'Content-Type': 'application/octet-stream',
               'Accept-Charset': 'UTF-8'
           },
           'body': {
               'errorMessage': self.messages
           }
       }
       return json.dumps(response)

APIGateway + Lambda + Pythonでステータスコードを変更するには、例外を発生させなければならないので、認証失敗のUnAuthorizationErrorクラスを独自で作成します。
戻り値のresponseのstatusCodeを401に変更し、errorMessageを追加します。

serverless.yml

# Lambdaのイベントトリガーを設定
events:
  # トリガーとしてAPIGatewayを構築
  - http:
      # リソースを指定
      path: sample/test
      # メソッドを指定
      method: post
      # 統合サービスをLambdaにする
      integration: lambda
      # カスタムレスポンスの作成
      response:
        # ヘッダーとテンプレートの指定
        headers:
          Content-Type: "'application/octet-stream'"
        template: $input.path('$')
        # カスタムレスポンスコードの設定
        statusCodes:
            # デフォルトのステータスコード
            200:
                pattern: ''
            # カスタムステータスコード
            401:
                pattern: '.*"statusCode": 401,.*'
                template: $input.path("$.errorMessage")
                headers:
                  Content-Type: "'application/octet-stream'"

serverless.ymlのhttpキーにresponseキーを追加し、さらにstatusCodesキーを追加します。
そのキーにステータスコードを追加します。patternキーの値に、レスポンスデータ(Lambdaからの戻り値)に含まれる文字列の正規表現を記入します。
デプロイ後に、作成されたAPIに認証エラーが起こるようにPOSTリクエストを送信します。

送信した情報
header
  Host:XXXXXXXXX.execute-api.ap-northeast-1.amazonaws.com
  Accept-Charset:utf-8
  Authorization:testtest
  Content-Type:application/octet-stream
body
  {"test":"test"}

クライアントツールで以下の画像のように表示されれば成功です。

DynamoDBの構築

Serverless Frameworkでは、serverless.ymlのResourceキーでDynamoDBやS3などのリーソースの設定を行うことができる。

テーブル作成

テーブル名がTableName、パーティションキー（ハッシュキー）が文字列型id、ソートキー（レンジキー）が文字列型のnameであるテーブルを作成します。

serverless.yml

# リソースの構築
resources:
  Resources:
    # DynamoDBの構築
    DynamoDbTable:
      Type: 'AWS::DynamoDB::Table'
      Properties:
        # キーの型を指定
        AttributeDefinitions:
          -
            AttributeName: id
            AttributeType: S
          -
            AttributeName: name
            AttributeType: S
        # キーの種類を指定（ハッシュorレンジキー）
        KeySchema:
          -
            AttributeName: id
            KeyType: HASH
          -
            AttributeName: name
            KeyType: RANGE
        # プロビジョニングするキャパシティーユニットの設定
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
        # テーブル名の指定
        TableName: TableName

Typeキーの値にDynamoDBの指定をします。PropertiesキーのAttributeDefinitionsの値に各キーの名前と型を指定し、KeySchemaの値に各キー名とキーの種類（HASH, RANGE）を指定します。
ProvisionedThroughputキーには、ReadCapacityUnitsに読み込みキャパシティーユニットを、WriteCapacityUnitsに書き込みキャパシティユニットを指定します。
最後にTableNameキーの値に作成するテーブル名を指定します。

デプロイ後に、AWSのコンソール画面からserverless.ymlで設定したテーブルが作成できているか確認してください。

テーブル操作

ここでは簡単にDynamoDBへのレコードの登録(put)、レコードの検索(query)、レコードの全件取得(scan)の関数を用意します。
boto3(AWS SDK for Python)を使用してテーブルを操作します。以下の3つの関数を作成します。

def put(id,name):
    """
    DynamoDBにレコードを登録する関数
    @Param id ハッシュキー
    @Param name レンジキー
    """
    table.put_item(
        Item = {
            "id" : id,
            "name" : name,
        }
    )


def query(id,name):
    """
    DynamoDBから検索する関数
    @Param id ハッシュキー
    @Param name レンジキー
    @return 検索結果
    """
    result = table.get_item(
        Key = {
            'id' : id,
            'name' : name,
        }
    )
    return result


def scan():
    """
    DynamoDBから全件検索する関数
    @return 検索結果
    """
    result = table.scan()
    return result

handler.pyを編集します。

handler.py

# coding: utf-8
"""
AWS Lambdaで実行するPythonコード

def hello(event,content) : Lambdaのhandler関数
def put(id,name) : DynamoDBにレコードを登録する関数
def query(id,name) : DynamoDBからレコードを検索する関数
def scan() : DynamoDBからレコードを全件取得する関数

class UnAuthorizationError(Exception) : 認証失敗の独自Exceptionクラス
    def __init__(self, code, messages) :  コンストラクタ
    def __str__(self) : 文字列変換メソッド
"""
import json
import logging
import boto3
import base64

# ログの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDBオブジェクトの作成
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('TableName')


def hello(event, context):
    """
    AWS Lambda Handler関数
    @Param event イベントデータ　APIGatewayからのデータ
    @Param content Lambdaのランタイムデータ
    @return APIGatewayのレスポンスデータ
    """
    # イベントデータの表示
    logger.info('headers:' + str(event['headers']))
    logger.info('body:' + str(event['body']))

    # 認証情報の取得
    authoriztion = str(event['headers']['Authorization'])
    # 独自認証。失敗した場合はExceptionを発生させ、カスタムレスポンスコード401を返す。
    if authoriztion != 'test':
        raise UnAuthorizationError(401,"errorMessage")

    # body部の取得
    body = json.loads(base64.b64decode(event['body']).decode('utf-8'))
    id = body['id']
    name = body['name']

    # DynamoDBにレコードの登録
    put(id,name)
    # DynamoDBから全件取得
    result = scan()

    # レスポンスデータの作成
    response = {
        "statusCode": 200,
        "body": json.dumps(result['Items'])
    }

    return response


def put(id,name):
    """
    DynamoDBにレコードを登録する関数
    @Param id ハッシュキー
    @Param name レンジキー
    """
    table.put_item(
        Item = {
            "id" : id,
            "name" : name,
        }
    )


def query(id,name):
    """
    DynamoDBから検索する関数
    @Param id ハッシュキー
    @Param name レンジキー
    @return 検索結果
    """
    result = table.get_item(
        Key = {
            'id' : id,
            'name' : name,
        }
    )
    return result


def scan():
    """
    DynamoDBから全件検索する関数
    @return 検索結果
    """
    result = table.scan()
    return result



class UnAuthorizationError(Exception):
    """
    認証失敗の独自Exceptionクラス
    @extends Exceptionクラスを継承
    """
    def __init__(self, code, data):
        """
        コンストラクタ
        @Param code レスポンスコード
        @Param data レスポンスデータ
        """
        self.code = code
        self.data = data

    def __str__(self):
        """
        文字列変換メソッド
        """
        response = {
           'status': 'HTTP/1.1 401 Unauthorized',
           'statusCode': self.code,
           'headers': {
               'Date': '2018/07/26 18:27:30',
               'Content-Type': 'application/octet-stream',
               'Accept-Charset': 'UTF8'
           },
           'body': {
               'result': self.code,
               'data': self.data
           }
        }
        return json.dumps(response)

リクエストのbody部からidとnameを取得しDynamoDBに登録し、DynamoDBからレコードを全件取得しレスポンスとして返します。
デプロイし、リクエストのbody部に{"id":"testid","name":"testname"}のようなJSONデータを指定して　POSTしてみてください。以下のようなレスポンスが帰ってきます。

statusCode=200, body=[{"id": "sampleid3", "name": "samplename3"}, {"id": "sampleid", "name": "samplename"}, {"id": "sampleid2", "name": "samplename2"}]}

body=以下がDynamoDBのレコードになります。

まとめ

Serverless Frameworkを使用したAWS Lambda、APIGateway、DynemoDBの構築からデプロイまでを行いました。
この3つのサービスで簡単なREST APIを作成することができます。
公式ページに今回紹介した以外の設定方法が載っていますので、必要な設定を追加してみてください。
以下に、最終的なserverless.ymlファイルとhandler.pyファイルを載せます。

serverless.yml

# サービス名を指定
service: slf-sample 

# 使用するクラウドサービス(AWS)と言語(Pytho3.6)を指定
provider:
  name: aws
  runtime: python3.6

# ステージを指定。開発と本番環境を切り分けることができます。
  stage: dev
# デプロイするリージョンを指定
  region: ap-northeast-1
# Lambdaに適用するIAM Roleを指定
  role: arn:aws:iam::XXXXXXXXX:role/lambda_basic_execution

# Lambdaを構築
functions:
  # 関数名を指定
  hello:
    # handler関数を指定
    handler: handler.hello
    timeout: 200
    # Lambdaのイベントトリガーを設定
    events:
      # トリガーとしてAPIGatewayを構築
      - http:
          # リソースを指定
          path: sample/test
          # メソッドを指定
          method: post
          # 統合サービスをLambdaにする
          integration: lambda
          # カスタムリクエストの作成
          request:
            # カスタムテンプレートの作成
            template:
              application/octet-stream:
                '{"headers":{
                  #foreach($key in $input.params().header.keySet())
                  "$key": "$input.params().header.get($key)"#if($foreach.hasNext),#end
                  #end
                  },
                  "body": "$util.base64Encode($input.json(''$''))"
                  }'
          # カスタムレスポンスの作成
          response:
            # ヘッダーとテンプレートの指定
            headers:
              Content-Type: "'application/octet-stream'"
            template: $input.path('$')
            # カスタムレスポンスコードの設定
            statusCodes:
                # デフォルトのステータスコード
                200:
                    pattern: ''
                # カスタムステータスコード
                401:
                    pattern: '.*"statusCode": 401,.*'
                    template: $input.path("$.errorMessage")
                    headers:
                      Content-Type: "'application/octet-stream'"
# リソースの構築
resources:
  Resources:
    # DynamoDBの構築
    DynamoDbTable:
      Type: 'AWS::DynamoDB::Table'
      Properties:
        # キーの型を指定
        AttributeDefinitions:
          -
            AttributeName: id
            AttributeType: S
          -
            AttributeName: name
            AttributeType: S
        # キーの種類を指定（ハッシュorレンジキー）
        KeySchema:
          -
            AttributeName: id
            KeyType: HASH
          -
            AttributeName: name
            KeyType: RANGE
        # プロビジョニングするキャパシティーユニットの設定
        ProvisionedThroughput:
          ReadCapacityUnits: 1
          WriteCapacityUnits: 1
        # テーブル名の指定
        TableName: TableName

handler.py

# coding: utf-8
"""
AWS Lambdaで実行するPythonコード

def hello(event,content) : Lambdaのhandler関数
def put(id,name) : DynamoDBにレコードを登録する関数
def query(id,name) : DynamoDBからレコードを検索する関数
def scan() : DynamoDBからレコードを全件取得する関数

class UnAuthorizationError(Exception) : 認証失敗の独自Exceptionクラス
    def __init__(self, code, messages) :  コンストラクタ
    def __str__(self) : 文字列変換メソッド
"""
import json
import logging
import boto3
import base64

# ログの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDBオブジェクトの作成
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('TableName')


def hello(event, context):
    """
    AWS Lambda Handler関数
    @Param event イベントデータ　APIGatewayからのデータ
    @Param content Lambdaのランタイムデータ
    @return APIGatewayのレスポンスデータ
    """
    # イベントデータの表示
    logger.info('headers:' + str(event['headers']))
    logger.info('body:' + str(event['body']))

    # 認証情報の取得
    authoriztion = str(event['headers']['Authorization'])
    # 独自認証。失敗した場合はExceptionを発生させ、カスタムレスポンスコード401を返す。
    if authoriztion != 'test':
        raise UnAuthorizationError(401,"errorMessage")

    # body部の取得
    body = json.loads(base64.b64decode(event['body']).decode('utf-8'))
    id = body['id']
    name = body['name']

    # DynamoDBにレコードの登録
    put(id,name)
    # DynamoDBから全件取得
    result = scan()

    # レスポンスデータの作成
    response = {
        "statusCode": 200,
        "body": json.dumps(result['Items'])
    }

    return response


def put(id,name):
    """
    DynamoDBにレコードを登録する関数
    @Param id ハッシュキー
    @Param name レンジキー
    """
    table.put_item(
        Item = {
            "id" : id,
            "name" : name,
        }
    )


def query(id,name):
    """
    DynamoDBから検索する関数
    @Param id ハッシュキー
    @Param name レンジキー
    @return 検索結果
    """
    result = table.get_item(
        Key = {
            'id' : id,
            'name' : name,
        }
    )
    return result


def scan():
    """
    DynamoDBから全件検索する関数
    @return 検索結果
    """
    result = table.scan()
    return result


class UnAuthorizationError(Exception):
    """
    認証失敗の独自Exceptionクラス
    @extends Exceptionクラスを継承
    """
    def __init__(self, code, data):
        """
        コンストラクタ
        @Param code レスポンスコード
        @Param data レスポンスデータ
        """
        self.code = code
        self.data = data

    def __str__(self):
        """
        文字列変換メソッド
        """
        response = {
           'status': 'HTTP/1.1 401 Unauthorized',
           'statusCode': self.code,
           'headers': {
               'Date': '2018/07/26 18:27:30',
               'Content-Type': 'application/octet-stream',
               'Accept-Charset': 'UTF8'
           },
           'body': {
               'result': self.code,
               'data': self.data
           }
        }
        return json.dumps(response)