API GatewayとLambdaの備忘

API Gatewayについて触ったことがなかったため、とある構成を模倣してテストを行った備忘

やりたいこと

"API①を叩く何か"がAPI①を叩いてメンバーIDの情報を取ろうとする ↓ API①がLambda①を呼び出す ↓ Lambda①がコードの中でAPI②を叩く ↓ API②がメンバーIDの情報を返す

前提条件

"API①を叩く何か"は直接API②を叩きたくない なので中継としてAPI①、Lambda①を作成してAPI②を叩き、情報を取ってくる API②はアクセス元を絞っているため、Lambda①からのアクセスIPは固定する必要がある →Lambda①をVPC内に配置し、固定IPを振ったNATから外に出るようにする Lambda②では、認証キーの認証が行われる リクエスト、レスポンスの変換は特になし IAM認証は省きます

手順

API①側のためのVPC①を作成 VPC①にパブリックサブネット①を作成（0.0.0.0/0 → IGW ルート） VPC①にプライベートサブネット①を作成（0.0.0.0/0 → NAT GW ルート） VPC①にIGWをアタッチ パブリックサブネット①にNAT GWを作成

API②側のためのVPC②を作成 VPC②にプライベートサブネット②を作成

Lambda① を VPC①のプライベートサブネット① に作成（SG/NACL設定含む）

以下参考Lambda

バリデーションチェック、適当な認証処理、メンバーリストがある(DBを作るのが面倒だったため、テスト用に)

import json
import base64

def lambda_handler(event, context):
    """
    POSTリクエストのJSONボディからMemberIDを受け取り、
    バリデーションとIDごとの情報取得を行うLambda関数
    """

    # 応答データとステータスコードを初期化
    response_data = {
        'success': False,
        'errorMessage': '内部サーバーエラー',
        'responseWrapper': {}
    }
    status_code = 500

    try:
        # --- 認証処理 ---
        headers = event.get('headers', {})
        authorization_header = headers.get('Authorization') 

        # Authorizationヘッダーが存在しない、または値が'12345'でない場合
        if not authorization_header or authorization_header != '12345':
            response_data['errorMessage'] = '認証エラー: Authorization ヘッダーが不正です。'
            status_code = 401
            return {
                'statusCode': status_code,
                'body': json.dumps(response_data, ensure_ascii=False)
           }

        # リクエストボディをデコード
        body_string = event['body']
        if event.get('isBase64Encoded', False):
            try:
                body_string = base64.b64decode(body_string).decode('utf-8')
            except Exception:
                response_data['errorMessage'] = 'リクエストボディが不正なBase64形式です。'
                status_code = 400
                return {
                    'statusCode': status_code,
                    'body': json.dumps(response_data, ensure_ascii=False)
                }

        body = json.loads(body_string)
        member_id = body.get('inputs', {}).get('MemberID')

        # ① MemberIDのバリデーションチェック
        if not member_id:
            response_data['errorMessage'] = 'MemberIDがリクエストに含まれていません。'
            status_code = 400
            return {
                'statusCode': status_code,
                'body': json.dumps(response_data, ensure_ascii=False)
            }

        if not (isinstance(member_id, str) and len(member_id) == 4 and member_id.isdigit()):
            response_data['errorMessage'] = 'MemberIDは4桁の数字文字列で指定してください。'
            status_code = 400
            return {
                'statusCode': status_code,
                'body': json.dumps(response_data, ensure_ascii=False)
            }

        valid_ids = ['0000', '0001', '0002', '0003']
        if member_id not in valid_ids:
            response_data['errorMessage'] = '無効なMemberIDです。利用可能な値は0000〜0003です。'
            status_code = 400
            return {
                'statusCode': status_code,
                'body': json.dumps(response_data, ensure_ascii=False)
            }

        # ② IDごとの情報定義
        members_info = {
            "0000": {"points": 133, "point_expiry_date": "20251212"},
            "0001": {"points": 1333, "point_expiry_date": "20251112"},
            "0002": {"points": 1, "point_expiry_date": "20250912"},
            "0003": {"points": 9999, "point_expiry_date": "20251012"}
        }

        # ③ IDに対応する情報を取得
        member_info = members_info.get(member_id)

        # ④ 成功時のレスポンスを組み立て
        response_data = {
            'success': True,
            'errorMessage': None,
            'responseWrapper': {
                'memberId': member_id,
                'points': member_info['points'],
                'point_expiry_date': member_info['point_expiry_date']
            }
        }
        status_code = 200

    except json.JSONDecodeError:
        response_data['errorMessage'] = 'リクエストボディが不正なJSON形式です。'
        status_code = 400
    except Exception as e:
        # その他の予期せぬエラー
        response_data['errorMessage'] = f'予期しないエラーが発生しました: {str(e)}'
        status_code = 500

    # 最終的なレスポンスを返す
    return {
        'statusCode': status_code,
        'body': json.dumps(response_data, ensure_ascii=False)
    }

API①をPOSTで作成し、Lambda①と統合してデプロイ 必要に応じてヘッダーの必須条件を設定する(認証キーが絶対必要とか。今回はなし)

Lambda② を VPC②のプライベートサブネット② に作成

API②をPOSTで作成し、Lambda②と統合、リソースポリシーで NAT GWのEIPのみ を許可してデプロイ

以下参考Lambda

API②のエンドポイントと認証キーはSecretManager管理なのでそこから取ってくる

import os
import json
import base64
import urllib.request
import urllib.error
import boto3

# 環境変数
URL_SECRET_NAME = os.environ["URL_SECRET_NAME"]
TOKEN_SECRET_NAME = os.environ["TOKEN_SECRET_NAME"] 


def get_secret(secret_name: str) -> str:
    #Secrets Manager から文字列を取得
    sm = boto3.client("secretsmanager")
    res = sm.get_secret_value(SecretId=secret_name)
    return res["SecretString"]


def lambda_handler(event, context):
    try:
        #Secrets取得
        url = get_secret(URL_SECRET_NAME)
        token = get_secret(TOKEN_SECRET_NAME)

        #リクエストボディ
        body = event.get("body")
        if event.get("isBase64Encoded"):
            body = base64.b64decode(body).decode("utf-8")

        # body が dict の場合は JSON に変換
        if isinstance(body, dict):
            body_str = json.dumps(body)
        else:
            body_str = body

        #ヘッダー 
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}"
        }

        #API呼び出し
        req = urllib.request.Request(
            url=url,
            data=body_str.encode("utf-8") if body_str else None,
            headers=headers,
            method="POST"
        )

        with urllib.request.urlopen(req, timeout=8) as resp:
            return {
                "statusCode": resp.getcode(),
                "headers": {"Content-Type": "application/json"},
                "body": resp.read().decode("utf-8")
            }

    except urllib.error.HTTPError as e:
        # APIが 4xx / 5xx を返したとき
        return {
            "statusCode": e.code,
            "body": e.read().decode("utf-8", errors="replace")
        }
    except Exception as e:
        # Secrets取得失敗、JSON処理失敗、ネットワークエラーなど
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)}, ensure_ascii=False)
        }

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": "*",
            "Action": "execute-api:Invoke",
            "Resource": "【API②】"
        }
    ]
}

実行

PostmanからAPIを叩いてみる ヘッダーに認証コードを入れる

返ってきた、OK