前提条件

ここで扱う API Gateway と Lambda はカスタム(非プロキシ)統合であることを前提に話を進めます

プロキシ統合とカスタム統合についての詳しい説明は省きますが、簡単に説明すると、プロキシ統合は API Gateway に飛んできたリクエストをそのまま Lambda 関数に event として入力します。出力も同様に Lambda 関数からの結果を返します。カスタム統合は、飛んできたリクエストをマッピングテンプレート等で event に入力するデータを変換し、Lambda 関数を呼び出す形で実行します。出力も同様にマッピングテンプレート等を使用して変換したものをレスポンスとして返します。要は、プロキシ統合を使うと、Lambda だけで完結するが、カスタム統合は API Gateway の設定(マッピングテンプレート等) が必要ということです

ちなみに、AWS のドキュメントではプロキシ統合を勧めています：

Lambda プロキシ統合は、1 つの Lambda 関数との合理化された統合設定をサポートしています。設定はシンプルで、既存の設定を破棄することなくバックエンドで拡張できます。このような理由から、Lambda 関数との統合を強くお勧めします

AWS | API ゲートウェイ API 統合タイプの選択

個人的にもプロキシ統合をオススメしますが、~~(色々あって)~~カスタム統合でのエラーハンドリングを考察していきます

例

ここでは、例として敵に何回攻撃したら倒せるかを求める Web API を作成します
プレイヤーの名前と敵のヒットポイントを入力したら、あらかじめ持っているデータからプレイヤーの攻撃力を参照して敵に何回攻撃したら倒せるかを計算して返します

簡単に Lambda 関数を実装すると以下のようになります：

lambda_function.py

import math

PLAYERS = {
    'Alice': 5,
    'Bob': 0
}


def lambda_handler(event, context):
    player_name, enemy_hp = event['name'], event['hp']

    return math.ceil(enemy_hp / PLAYERS[player_name])

しかし、このままではエラーが多く潜んでいることがわかります。次のエラーが考えられます：

event に name、hp が入っているか、正しい値が入っているか
PLAYERS にユーザーが登録されているか
ゼロ除算していないか

これらのエラーをハンドリングしながら関数をリファクタリングしていきます

想定されるエラーと HTTP レスポンスを照らし合わせる

コードの中で想定されるエラーと、そのエラーが起きた場合どの HTTP ステータスコードを返すか考えて処理を分割していきます。今回だと次のように整理できます：

lambda_function.py

import math

PLAYERS = {
    'Alice': 5,
    'Bob': 0
}


def lambda_handler(event, context):
    # ===========
    # 400 Bad Request
    player_name, enemy_hp = event['name'], event['hp']
    # ===========

    # ===========
    # 404 Not Found
    attack_point = PLAYERS[player_name]
    # ===========

    # ===========
    # 500 Internal Server Error
    attack_times = math.ceil(enemy_hp / attack_point)
    # ===========
    return attack_times

HTTP レスポンスに使う Exception を作成する

今のままでは、KeyError や ZeroDivisionError などのランタイムのエラーがレスポンスとして帰ってしまいます。なので、HTTP レスポンスに使う自作 Exception を作成していきます。

lambda_exceptions.py

import json


class LambdaException(Exception):
    def __init__(self, status_code: int, error_msg: str):
        self.status_code = status_code
        self.error_msg = error_msg

    def __str__(self):
        obj = {
            "statusCode": self.status_code,
            "errorMessage": self.error_msg
        }
        return json.dumps(obj)


class NotFoundException(LambdaException):
    def __init__(self, error_msg: str):
        super().__init__(404, error_msg)


class BadRequestException(LambdaException):
    def __init__(self, error_msg: str):
        super().__init__(400, error_msg)


class InternalServerErrorException(LambdaException):
    def __init__(self, error_msg: str):
        super().__init__(500, error_msg)

後半を見てもらえばわかるように、HTTP レスポンス用の Exception にしています。理由は API Gateway の統合レスポンスではステータスコード毎にマッピングする必要があるため、ステータスコードを固定することで typo や想定していないステータスコードを書かれることを防ぐためです。(設定していないステータスコードがきても 200 番(OK)になってしまう)

そうしたら、lambda_handler に追加していきます：

lambda_function.py

import math
import traceback

from lambda_exceptions import BadRequestException, NotFoundException, InternalServerErrorException

PLAYERS = {
    'Alice': 5,
    'Bob': 0
}


def lambda_handler(event, context):
    try:
        player_name, enemy_hp = event['name'], event['hp']
        if not isinstance(player_name, str) or not isinstance(enemys_hp, int):
            raise BadRequestException('name または hp の型が違います')
    except Exception as e:
        traceback.print_exc()
        raise BadRequestException('name と hp が必要です')

    try:
        attack_point = PLAYERS[player_name]
    except Exception as e:
        traceback.print_exc()
        raise NotFoundException('プレイヤーが見つかりませんでした')

    try:
        attack_times = math.ceil(enemy_hp / attack_point)
    except Exception as e:
        traceback.print_exc()
        raise InternalServerErrorException('ゼロ除算をしています')
    
    return attack_times

紹介のためにここでは Exception でキャッチしていますが、 Exception をキャッチするのは広すぎるので、想定したエラー (e.g. KeyError) のときにはそれに対応した例外を投げて、それ以外は全て 500 番等で処理する方が良いと思います。
後にも言いますが、このエラーメッセージはレスポンスメッセージとして返るものになるため、第三者に見られてもいい内容がベターです。

traceback はスタックトレースを出力するのに必要です

レスポンスのスタックトレースを消す

現在のままでは、エラーが起こる入力を投げると次のようなレスポンスが帰ってきます：

{
    "errorMessage": "{\"statusCode\": 400, \"errorMessage\": \"name \ま\た\は hp \の\型\が\違\い\ま\す\"}",
    "errorType": "BadRequestException",
    "stackTrace": [
        "  File \"/var/task/lambda_function.py\", line 14, in lambda_handler\n    raise BadRequestException('name または hp の型が違います')\n"
    ]
}

スタックトレースが返されており、見せる必要がないものを見せています。これの errorMessage のみを返すためにマッピングテンプレートを設定します

$input.path('$.errorMessage')

統合レスポンスはこんな感じになります：

処理部分とインターフェイス部分を意識する

ここでいう処理部分は、API Gateway からそのまま切り離しても問題がない関数の事を指します。逆にインターフェイス部分は上で作成した自作 Exception のように、Web API レスポンスに関係する部分のことを指します。

筆者はこのインターフェイス部分は最小限であることがよく、処理部分と分離することが望ましいと考えています

処理部分を集めたファイルを main.py として作成していきます：

main.py

import math

PLAYERS = {
    'Alice': 5,
    'Bob': 0
}


def unwrap_event(key_types: list, event: dict) -> list:
    results = []

    for k, t in key_types:
        if k not in event:
            raise KeyError(f'key が event にありません. key: {k}, event: {event}')
        if not isinstance(event[k], t):
            raise TypeError(f'event の値が想定した型ではありません. event[{k}]: {event[k]}, expected type: {t}')

        results.append(event[k])

    return results


def find_attack_point_of_player(name: str) -> int:
    if name not in PLAYERS:
        raise KeyError(f'入力したプレイヤー名が PLAYERS に登録されていません. name: {name}, PLAYERS: {PLAYERS}')
    return PLAYERS[name]


def seek_attack_times(hp: int, attack_point: int)-> int:
    return math.ceil(hp / attack_point)

ここでのポイントは

HTTP レスポンス用の Exception を使わない
外に出ないので、データの詳細を出力させる

ことです。処理部分ではステータスコードは必要なく、KeyError などの Python の例外処理を書き、エラー時にはデータの詳細をメッセージとして残します。このようにすることによって、ユニットテストが明確に書け、ログからバグ修正をスムーズに行うことができます

最終的な lambda_function.py は次のようになります：

lambda_function.py

import json
import traceback

from main import unwrap_event, find_attack_point_of_player, seek_attack_times
from lambda_exceptions import BadRequestException, NotFoundException, InternalServerErrorException


def lambda_handler(event, context):
    try:
        player_name, enemy_hp = unwrap_event([('name', str), ('hp', int)], event)
    except TypeError as e:
        traceback.print_exc()
        raise BadRequestException('name または hp の型が違います')
    except Exception as e:
        traceback.print_exc()
        raise BadRequestException('name と hp が必要です')
    
    try:
        attack_point = find_attack_point_of_player(player_name)
    except Exception as e:
        traceback.print_exc()
        raise NotFoundException('プレイヤーが見つかりませんでした')

    try:
        attack_times = seek_attack_times(enemy_hp, attack_point)
    except Exception as e:
        traceback.print_exc()
        raise InternalServerErrorException('ゼロ除算をしています')

    return f'{player_name} は {attack_times} 回攻撃する必要があります'

さいごに

長くなりましたが、筆者のカスタム統合での Lambda + API Gateway のエラーハンドリングに関する案を紹介しました
間違いや、より良い方法があればご教授いただければと思います