WebAPI 実装ガイド

本カタログで WebAPI を実装するプラクティスを記載したドキュメントです。ポイントは以下2点です。

• WebAPI の主要機能は AWS 公式ライブラリ Powertools for AWS Lambda (Python) を利用
• 内部アーキテクチャとして Lambdalith 構成を採用
  • API ごとに Lambda リソースを分離するより、パフォーマンスとと運用便宜性に優れる
  • 従来の WebAPI フレームワークの開発構成を踏襲できる

ユーザー情報を登録する API の実装

ユーザー情報の入力を受け、ユーザーデータを新規登録する API を実装します。

API 仕様

以下のような仕様のサマリーを作成します。

• 正常系

  [Request sample]
  POST /users
  {
    "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }
  
  [Response sample 1]
  200 OK
  {
    "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }

• 異常系（name未指定）

  [Request sample]
  POST /users
  {
    # "name": "Shigeo",
    "country": "Japan",
    "age": 30
  }
  
  [Response sample]
  400 Bad Request
  {
    "message": "Missing name. name is required query parameter."
  }

大まかな処理の流れ

Lambdalith の推奨構成として、大きく以下４つのレイヤーで構成され、上から順に呼ばれていきます。

レイヤー名	サマリー
API Routing	マッピングされた Action を呼び出す
Action	入力チェック、Service の呼び出しを行う
Service	API のメインロジック・ビジネスロジック
Dao	データベースアクセス処理・クエリー実行

他にも、AWS SDK 実装を分離した aws/、外部ライブラリを分離した lib/、データのモデリングやスキーマを定義する models/ など、必要に応じて適宜ディレクトリを構成してください。

API Routing 実装

リクエストを受け付けると、Lambda 関数のエントリーポイント（ src/sample_api/index.ts ）にリクエスト情報が入った event オブジェクトが渡されます。Powertools から提供されるデコレータを利用して以下のように適切な Action に振り分けます。

実装例

• src/sample_api/index.ts

from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.exceptions import BadRequestError

# app をインスタンス化
app = APIGatewayHttpResolver()

# Lambda 関数のエントリーポイント
def handler(event, context):
    return app.resolve(event, context)

# POST /users をこの関数にマッピング
@app.post("/users")
def post_users():
    event = app.current_event
    return CreateUsersAction().handle(http_api_event=event)

# 処理の中で BadRequestError が投げられたら、こちらの関数が呼ばれる
@app.exception_handler(BadRequestError)
def handle_400_bad_request_error(exception):
    return Response(status_code=400, content_type="application/json", body=json.dumps({"message": exception.msg}))

# 処理の中で意図しない Exception が投げられたら、こちらの関数が呼ばれる
@app.exception_handler(Exception)
def handle_500_internal_server_error(e: Exception):
  # こちらのサンプルコードはカタログ AMI に含まれています。

Action 実装

「Action」は、以下の役割を担当します。

• API 処理の「入り口」と「出口」となる部分
• 入力チェック、Service 実行、レスポンスデータ型の整理等を行う
• Controller 層と言われることもある

実装例

• src/sample_api/actions/create_users_action.py

import json
from aws_lambda_powertools.event_handler.exceptions import BadRequestError
from sample_api.services.users_service import UsersService

class CreateUsersAction:

    def handle(self, http_api_event):
        # リクエストボディを取り出す
        req_body = json.loads(http_api_event.body)

        # リクエストボディの項目を取り出す
        name = req_body.get('name')
        country = req_body.get('country')
        age = req_body.get('age')

        # 入力チェックを行う
        self.validate_required_params(name)
        # 他にも、age が numeric 型であるか、そもそも body が None になっていないかの確認が必要

        # Service のインスタンス化と実行
        service = UsersService()
        service.create_users(name, country, age)

        # レスポンス情報の組み立てと返却
        return {
            'statusCode': 200,
            'body': {
                'message': "Successfully created user data."
            }
        }

    def validate_required_params(self, name):
        if name is None:
            raise BadRequestError('Missing name. name is required query parameter.')

    def validate_body_is_not_empty(self, http_api_event):
      # こちらのサンプルコードはカタログ AMI に含まれています。

    def validate_numeric_params(self, age):
      # こちらのサンプルコードはカタログ AMI に含まれています。

    def is_numeric(self, val):
      # こちらのサンプルコードはカタログ AMI に含まれています。

Service 実装

「Service」は、API のメインロジック、ビジネスロジックを担当する部分です。

実装例

• src/sample_api/services/users_service.py

from aws_lambda_powertools import Logger
from sample_api.daos.users_dao import UsersDao

users_dao = UsersDao()

class UsersService:

    def create_users(self, name, country, age):
        # ユーザー作成のための処理を実装します
        # 例えば、ユニークな ID 採番ルールがある場合は、採番処理を実装します
        # 登録準備が整ったら、DAO を経由してデータベースにユーザー情報の書き込み処理を実施します
        new_users_record = users_dao.create_users(name, country, age)

        return new_users_record

DAO 実装

「DAO」は、データベースにアクセスし、クエリーを実行する処理を担当します。

実装例

• src/sample_api/actions/users_service.py

import os, psycopg
from psycopg.rows import dict_row


class UsersDao:
    def get_connection_info(self):
        # psycopg3 のコネクション作成に必要なパラメータ情報を読み込みます。
        # こちらのサンプルコードはカタログ AMI に含まれています。

    def create_user(self, name, country, age):
        try:
            # データベースのコネクションを作成します
            # 実際には、コネクションプールの仕組みを検討してください。
            conn_info = self.get_connection_info()
            conn = psycopg.connect(**conn_info)
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO users (name, country, age) VALUES (%s, %s, %s)",
                    (name, country, age)
                )
                conn.commit()
            conn.close()

動作確認

「ローカル実行によるデバッグとユニットテスト」を参照してください。