Help us understand the problem. What is going on with this article?

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

More than 1 year has passed since last update.

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をデプロイ

architect.png

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.png

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

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クラスを独自で作成します。
戻り値のresponsestatusCodeを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.ymlhttpキーに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"}

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

client.png

DynamoDBの構築

Serverless Frameworkでは、serverless.ymlResourceキーで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.png

テーブル操作

ここでは簡単に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部からidnameを取得し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::790499047482: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)

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした