はじめに

AWSのサーバーレスアーキテクチャを中心に設計・構築を担当するエンジニアです。これまで「ECS上でコンテナを動かす」スタイルで開発体験を最適化してきましたが、Lambda中心のアーキテクチャに移行する際、多くの開発者が「ローカルでいかにLambda環境を再現するか」という問いに嵌まりがちです。

私自身も当初は「ホットリロードが効かない」「イベントの再現が面倒」といった課題に対し、ローカル再現に心血を注いできました。しかし、本検証を通じて至った結論は、「ローカルでの再現」を目指すこと自体がアンチパターンであるということです。

この記事では、サーバーレス開発における「開発・検証・本番」の境界をどう設計し、それぞれのフェーズに最適な手段をどう割り当てるべきか、という問いに対する一つの回答を提示します。

レポジトリ：

前編

対象読者

AWS LambdaやAPI Gatewayを使ったサーバーレス開発に携わっているエンジニア
ローカルでのLambda再現ツール（SAM CLI、LocalStack、serverless-offline など）の導入を検討している方
サーバーレス開発の「DXをどう改善するか」という問いに向き合っているチームリード・スタッフエンジニア

用いた技術スタック

本記事の検証は以下の構成で実施しました。

言語: Python 3.12
開発用フレームワーク: Flask 3.x
テスト: pytest + moto（AWSサービスのモック）
ローカルAWSエミュレーション: LocalStack（補助利用）

前編の要約

前編では、AWS Lambda 開発で重要なのは「ローカルで Lambda 実行環境をどこまで再現するか」ではなく、開発・検証・本番それぞれで何を担保するかを切り分けることだと整理しました。

後編では、その考え方をこのリポジトリでどう実装しているかを具体的に示します。焦点は次の 3 点です。

Flask をなぜ採用しているのか
handler をどう薄く保っているのか
その構造をどうやって運用で崩れにくくしているのか

前編で扱った速度比較や各ツールの検証結果は背景として重要ですが、本稿では数値の再掲よりも、現行 repo で再現できる設計と運用に絞って説明します。

なぜ Flask なのか

このプロジェクトで Flask を使う理由は、Lambda を再現するためではありません。開発中の試行錯誤を速くするための HTTP アダプタとして使うためです。

重要なのは、Flask がビジネスロジックの実行主体ではなく、request を Lambda 互換の event dict に変換して既存の handler を呼ぶだけの薄い層になっていることです。

ローカル開発: Flask Request -> event dict -> lambda_handler
本番実行:     API Gateway Event -> lambda_handler

共通点:
handler が受け取る入力はどちらも event dict

この形にしておくと、ローカル開発では HTTP で素早く確認でき、本番では API Gateway から同じ handler をそのまま実行できます。Flask は便利な Web フレームワークですが、この repo における役割はあくまで 開発用アダプタ です。

FastAPI や ASGI を否定したいわけではありません。ただ、この repo の目的は「開発レイヤーを薄く保つこと」なので、必要十分な機能だけで済む Flask が合っている、というのが実務上の判断です。

現行 repo での層の分け方

この repo の中核は、HTTP と Lambda handler を明確に分離している点にあります。

api/
├── main.py
├── routes.yaml
├── adapters/
│   ├── event_builder.py
│   ├── response_mapper.py
│   ├── router.py
│   └── http_flask.py
└── functions/
    ├── common.py
    ├── calc/
    ├── async_task/
    ├── task_worker/
    ├── email_publisher/
    └── sns_mailer/

責務は次のように分かれています。

adapters/event_builder.py: Flask Request を API Gateway v2 風の event dict に変換
adapters/response_mapper.py: Lambda 形式レスポンスを Flask Response 向けに正規化
adapters/http_flask.py: route 解決と handler 実行を接続
functions/*: event を受けて実際のユースケースを処理
routes.yaml: パス、HTTP メソッド、handler の対応を一元管理

routes.yaml は単純なルートだけでなく、同一パスに対するメソッド別 handler も扱えます。

routes:
  - path: /calc
    methods: [POST]
    handler: api.functions.calc.lambda_handler:lambda_handler

  - path: /db
    methods_map:
      POST: api.functions.db.write:lambda_handler
      GET: api.functions.db.read:lambda_handler

Flask 側は handler 固有の実装をほとんど持ちません。実際の流れは次のとおりです。

def view_func():
    event = from_flask_request(request)
    result = handler(event, None)
    mapped = map_response(result)
    return _build_flask_response(mapped)

この構造のおかげで、「HTTP で試す」と「Lambda handler を直接呼ぶ」が同じ計算経路を通ります。

ローカルでの確認手順

この repo では Makefile ベースで確認できます。

make up
make dev
curl -X POST http://localhost:8000/calc \
  -H "Content-Type: application/json" \
  -d '{"x": 10, "y": 2, "op": "div"}'

ここで重要なのは、curl が直接ビジネスロジックを叩いているのではなく、

Flask Request
event_builder による event 化
lambda_handler(event, None)
response_mapper による HTTP 応答化

という流れを通っていることです。つまり、Flask 開発であっても handler の入口は Lambda 互換です。

検証レイヤーは 2 種類ある

この repo のテストは、単に「pytest がある」ではなく、目的の異なる 2 種類の検証手段に分かれています。

1. Unit テスト: fixture で event を生成して handler を直接叩く

unit テストでは Flask を使いません。conftest.py の fixture から API Gateway v2 形式の event を作り、handler を直接呼びます。

def test_calc_success(apigw_event):
    event = apigw_event(
        method="POST",
        path="/calc",
        body='{"x": 3, "y": 2, "op": "add"}',
    )
    response = lambda_handler(event, None)
    assert response["statusCode"] == 200

この方法の利点は、HTTP レイヤーに引きずられず、handler の入出力契約を高速に検証できることです。

実行は Makefile から行います。

make test

2. Console テスト: JSON ケースで AWS コンソール相当の確認をする

もう一つの検証が api/tests/console/cases/ 配下の JSON ドリブンなケースです。こちらは unit テストとは役割が違い、AWS コンソールで event JSON を流し込む感覚に近い確認をローカルで回すための仕組みです。

構成は次のようになっています。

api/tests/console/cases/
└── <event_type>/
    └── <usecase>/
        ├── config.json
        ├── <case>.event.json
        └── <case>.expected.json

全ケース実行と個別実行は Makefile から行えます。

make run-all-cases
make run-all-cases FUNCTION=calc
make run-case FUNCTION=calc CASE=add_positives

この 2 層を分けているのがポイントです。unit テストはロジックを速く固めるため、console テストは event 入力の見通しを良くするためにあります。どちらも handler を直接呼ぶので、Flask に依存しません。

薄い handler を保つ実装パターン

この repo で一貫しているのは、Lambda handler を 薄いディスパッチャー に保つことです。

def lambda_handler(event: dict, context) -> dict:
    return safe_execute(execute, event)

やることは 2 つだけです。

execute(event) に処理を委譲する
例外を safe_execute() で Lambda 形式レスポンスへ変換する

実際の calc はこの形にかなり忠実です。

def execute(event: dict[str, Any]) -> dict[str, Any]:
    data = validate(event)

    x = data["x"]
    y = data["y"]
    op = data["op"]

    if op == "div" and y == 0:
        raise BadRequest("division by zero")

    result = OPERATIONS[op](x, y)
    return success({"result": result})


def lambda_handler(event: dict[str, Any], context: Any | None) -> dict[str, Any]:
    return safe_execute(execute, event)

このパターンの利点は明確です。

入力検証は execute() 側に閉じ込められる
Flask / pytest / AWS Lambda のどこから呼んでも入口が同じ
400 と 500 の責務分担を common.py に寄せられる

バリデーションで重視していること

特に API Gateway 経由の handler では、parse_body() と型チェックを厳密に扱うのが重要です。

parsed = parse_body(event)
if parsed["_type"] == "empty":
    raise BadRequest("request body is required")
if parsed["_type"] == "invalid":
    raise BadRequest("request body must be valid JSON")

body = parsed["data"]
if not isinstance(body, dict):
    raise BadRequest("request body must be a JSON object")

さらに数値入力では bool を先に弾きます。Python では bool が int のサブクラスなので、ここを曖昧にすると簡単にバグになります。

記事として伝えたいのは、「Lambda handler は薄く」のスローガンではなく、薄く保つための具体手順が既にコード化されているという点です。

非同期チェーンをどう足しているか

この repo には、同期 API を非同期ワークフローへ変換する実例があります。

API Gateway
  -> async_task
  -> SQS task-queue
  -> task_worker
     -> S3 保存
     -> SQS email-queue
  -> email_publisher
  -> SNS
  -> sns_mailer
  -> Mailhog

入口になる async_task は、重い処理を自分で抱え込みません。受け取ったリクエストを検証し、後続ワーカーに渡すためのタスクを SQS に積むだけです。

def execute(event: dict[str, Any]) -> dict[str, Any]:
    data = validate(event)
    body = json.dumps(data)

    resp = _sqs_client().send_message(
        QueueUrl=_TASK_QUEUE_URL,
        MessageBody=body,
    )
    return success({"message_id": resp["MessageId"]})

入力は現在の実装に合わせて次の形です。

{
  "key": "docs/document.txt",
  "content": "hello world",
  "email": "user@example.com",
  "subject": "Task ready",
  "message": "Your document was saved."
}

この設計の利点は、API の責務を「受け付け」に限定できることです。S3 保存やメール通知は後続 Lambda に分割されるので、

API は応答を早く返せる
再実行単位を細かくできる
ワークフローのどこで失敗したかを切り分けやすい

という実務上の利点があります。

この repo での確認手順

make up
make run-all-cases FUNCTION=async_task
make ci

ここでの make ci はデプロイではなく、ローカルで維持すべき品質ゲートを通すためのコマンドです。

ベストプラクティスをどう維持しているか

設計原則は、文章で掲げるだけではすぐ崩れます。この repo では、崩れにくくする仕組みを複数重ねています。

1. instructions による明文化

.github/instructions/ には、Python 実装、Lambda 境界、テスト設計のルールが分かれています。これにより、開発時に「何をどこで守るべきか」を読み物として参照できます。

2. routes.yaml による API 境界の固定

ルート定義を Flask 側に散らさず、routes.yaml に集約しているため、HTTP エンドポイントと handler の対応が追いやすくなっています。複数メソッドを同じパスで分けるケースも見通しよく扱えます。

3. Makefile による確認手順の固定

この repo で実際に使うターゲットは次のとおりです。

make up
make dev
make test
make run-all-cases
make ci

make ci の中身も現行 repo に合わせて明確です。

ci: lint format test run-all-cases

つまり、この repo での CI は次の責務を持っています。

lint: ruff check --fix
format: ruff format
test: unit テスト
run-all-cases: JSON ドリブンな console テスト

「SAM ビルドまで自動で担保している」わけではありません。ここを盛って書かないことが、公開記事では重要です。

4. テスト観点を揃えやすい構造

すべての handler が完全に同じ件数のテストを持っているわけではありません。ただし、少なくとも API Gateway 系 handler では次の観点を揃えやすい構造になっています。

正常系
必須フィールド欠落
不正型
bool 混入
非オブジェクト JSON
無効 JSON

「自動強制」と言い切るより、逸脱が見つけやすく、レビューしやすい構造を整えていると表現する方が現実に即しています。

実装から運用までの一貫性

このプロジェクトの価値は、Flask を使っていること自体ではありません。開発・検証・本番の境界を分けながら、handler の入口を共通化していることにあります。

開発では Flask から event dict を作る
unit テストでは fixture から event dict を作る
console テストでは JSON ケースから event dict を作る
本番では AWS が event dict を渡す

どこから呼ばれても、handler 側は event を受けて同じルールで処理する。この一貫性があるから、ローカル開発の速さと本番運用の安心感を両立できます。

Lambda 開発で本当に守りたいのは「ローカルでどこまで本番を真似るか」ではなく、「どこを共通化し、どこを分離するか」です。この repo は、その判断をコードと運用に落とし込んだ一例になっています。

まとめ

本稿で伝えたかったのは、次の 3 点です。

Flask は Lambda の代用品ではなく、開発用アダプタとして使うと強い。
handler は safe_execute(execute, event) を軸に薄く保つと、呼び出し元を問わず同じ実装を使える。
instructions、routes.yaml、Makefile、unit テスト、console テストを組み合わせると、その構造をチームで維持しやすくなる。

もしチームで導入するなら、最初に整えるべきなのは「ローカル再現環境」よりも、

event を共通入口にする設計
handler を薄く保つルール
make test と make ci を中心にした確認手順

の 3 つです。ここが揃えば、開発体験と本番品質は対立しにくくなります。