SQS のキューにメッセージが追加されたときに起動する AWS Lambda (Python 3.13) のスクリプトを、 AWS アカウント不要・完全ローカルの Docker Compose 環境で検証するためのツールです。
使い方は極めて簡単です:
lambda/handler.pyの中身を検証したいスクリプトに差し替える- キューにメッセージを送る
- ログで結果を確認する
さらに、以下のモックを内蔵しています。
- Python から呼び出す Java サーブレットの「呼び出しの書式・プロトコル・HTTP メソッド」を 簡単に確認できるモックサーブレット
- 暗号化された API キーなどを取得できるモック SSM Parameter Store
(
ssm.get_parameter(...)を本物どおりに書けます)
- 全体構成
- 前提条件
- クイックスタート (5 分で動かす)
- 検証したい Python スクリプトの差し替え方
- キューへのメッセージ送信方法
- サーブレット呼び出しの書式・プロトコル・HTTP メソッドの確認方法
- servlet_client の使い方 (Python からサーブレットを呼ぶ)
- Parameter Store から暗号化された値を取得する (モック SSM)
- 環境変数をテスト用に差し込む (インジェクション)
- SQS を通さず Lambda を直接起動する (デバッグ用)
- Lambda が受け取る SQS イベントの形式
- 本物の Java サーブレットに接続したい場合
- 追加の pip ライブラリを使いたい場合
- よく使うコマンド一覧
- ファイル構成
- トラブルシューティング
- 本物の AWS との違い (制限事項)
docker compose run --rm send
(message.json の内容を送信)
│
▼
┌──────────────┐ ポーリング ┌──────────────┐ SQSイベント ┌──────────────────────┐
│ ElasticMQ │◀────────────│ trigger │───────────────▶│ Lambda コンテナ │
│ (SQS互換) │ │ (SQS→Lambda │ (HTTP) │ Python 3.13 │
│ :9324 API │ │ トリガー) │ │ lambda/handler.py │
└──────────────┘ └──────────────┘ │ :9000 直接起動用 │
└────┬────────────┬────┘
HTTP │ │ SSM API
(GET/POST/PUT/...) │ │ (boto3)
▼ ▼
┌──────────────────────┐ ┌──────────────────────┐
│ mock-servlet │ │ mock-ssm │
│ (Javaサーブレットの代役)│ │ (Parameter Storeの代役)│
│ :8000 │ │ :8082 │
│ ・全リクエストを記録 │ │ ・暗号化パラメータを │
│ ・内容をエコーバック │ │ 復号して返す │
└──────────────────────┘ └──────────────────────┘
| コンテナ | 役割 |
|---|---|
| elasticmq | SQS 互換のキューサーバー。起動時に lambda-queue というキューが自動作成される。状態は docker compose run --rm status で確認できる |
| lambda | AWS 公式の Lambda Python 3.13 イメージ。lambda/handler.py がそのまま実行される |
| trigger | キューをポーリングし、メッセージを本物の AWS と同じ「SQS イベント形式」に包んで Lambda を起動する (AWS のイベントソースマッピング相当)。成功したらメッセージを削除、失敗したら残す (=再実行される) |
| mock-servlet | Java サーブレットの代役。届いたリクエストの HTTP メソッド・パス・プロトコル・ヘッダー・ボディをすべて記録し、そのままエコーバックする |
| mock-ssm | SSM Parameter Store の代役。本物と同じ AWS API プロトコルを話すので ssm.get_parameter(...) をそのまま書ける。値は mock-ssm/parameters.json を編集するだけで登録できる |
| send | テストメッセージ送信ツール (常駐しない)。docker compose run --rm send で使う |
| status | キュー状態確認ツール (常駐しない)。docker compose run --rm status で使う |
- Docker Desktop がインストール済みで起動していること (Windows / Mac / Linux いずれも可)
- 初回のみインターネット接続 (イメージのダウンロードに必要)
- 使用ポート:
9324900080008082が空いていること
AWS アカウント・AWS CLI・Python のローカルインストールは 一切不要 です。
このフォルダでターミナル (PowerShell など) を開いて実行します。
docker compose up -d --build初回はイメージのダウンロードとビルドで数分かかります。2 回目以降は数秒で起動します。
起動確認:
docker compose pselasticmq lambda trigger mock-servlet mock-ssm の 5 つが Up になっていれば OK です。
docker compose run --rm sendmessage.json の内容がキューに送信されます。以下のように表示されれば成功です:
キューにメッセージを送信しました
MessageId: xxxxxxxx-xxxx-....
本文 : { "user_id": 123, ... }
Lambda の実行ログ:
docker compose logs lambdaサーブレットが受け取った内容 (メソッド・プロトコル・ヘッダー・ボディ):
docker compose logs mock-servlet全コンテナのログを流れで見たい場合 (リアルタイム監視、Ctrl+C で抜ける):
docker compose logs -fdocker compose down編集するのは lambda/handler.py の 1 ファイルだけです。
-
lambda/handler.pyをエディタで開き、中身を検証したいスクリプトに書き換える- 入口関数は本物の Lambda と同じ
lambda_handler(event, context) eventには SQS イベント (後述の形式) がそのまま入ってくる
- 入口関数は本物の Lambda と同じ
-
変更を反映する (数秒で完了):
docker compose restart lambda
-
メッセージを送って動作確認:
docker compose run --rm send docker compose logs lambda
なぜ restart が必要?
lambda/フォルダはコンテナにマウントされているのでファイル自体は即時反映されますが、 Lambda ランタイムは一度読み込んだコードをメモリに保持するため、再読込に restart が必要です。 restart は数秒で終わります。
import json
from servlet_client import ServletClient
servlet = ServletClient()
def lambda_handler(event, context):
for record in event.get("Records", []):
body = record["body"] # SQS メッセージ本文 (文字列)
data = json.loads(body) # JSON なら dict に変換
# ここに検証したい処理を書く
res = servlet.post_json("/api/sample", data)
print(res.status, res.text)
return {"statusCode": 200}print()した内容はすべてdocker compose logs lambdaで見られます- ハンドラー内で例外が発生すると trigger が「エラー」と判定し、メッセージは削除されず 約 30 秒後 (可視性タイムアウト後) に自動で再実行されます — これも本物の SQS トリガーと同じ挙動です
-
リポジトリ直下の
message.jsonを送りたい内容に書き換える (JSON でなくてもよい。中身がそのまま SQS メッセージ本文になる) -
送信:
docker compose run --rm send
ファイル経由なので、シェルの引用符エスケープ問題が一切発生しません。日本語も改行もそのまま送れます。
docker compose run --rm send "テストメッセージです"注意 (PowerShell): 引数に JSON を直接書くと PowerShell が
"を除去してしまうことがあります。 JSON を送りたい場合は方法 1 (message.json) を使うのが確実です。
docker compose run --rm send "1件目"
docker compose run --rm send "2件目"
docker compose run --rm send "3件目"本物の SQS と同様、複数メッセージは最大 10 件ずつまとめて 1 回の Lambda 起動
(Records 配列に複数件) に入ることがあります。
docker compose run --rm statusキュー内の「処理待ち」「処理中 (配信済み)」「遅延中」のメッセージ数が表示されます。
==================================================
キュー名 : lambda-queue
URL : http://elasticmq:9324/000000000000/lambda-queue
--------------------------------------------------
処理待ち : 0 件
処理中 (配信済み) : 0 件
遅延中 : 0 件
==================================================
キューは空です。
Python (Lambda) から Java サーブレットをどう呼び出しているか (HTTP メソッド・URL パス・クエリ・プロトコルバージョン・ヘッダー・ボディの書式) を、 モックサーブレットがすべて自動記録します。確認方法は 3 通りあります。
docker compose logs mock-servletリクエストが届くたびに、次のような詳細ログが出力されます:
============================================================
受信リクエスト #1 (2026-07-22T12:00:00)
メソッド : POST
パス : /api/sample
クエリ : {}
プロトコル : HTTP/1.1
ヘッダー :
Content-Type: application/json; charset=utf-8
Content-Length: 58
Host: mock-servlet:8000
...
ボディ(JSON):
{"user_id": 123, "name": "太郎", "action": "sample"}
============================================================
リアルタイムで監視したい場合は -f を付けます:
docker compose logs -f mock-servletモックサーブレットは直近 100 件の受信履歴をメモリに保持しています。 ブラウザで http://localhost:8000/__history を開くか、コマンドで:
# PowerShell
Invoke-RestMethod http://localhost:8000/__history | ConvertTo-Json -Depth 10
# または curl (curl.exe / Git Bash / Mac / Linux)
curl http://localhost:8000/__history応答例:
{
"count": 1,
"requests": [
{
"seq": 1,
"received_at": "2026-07-22T12:00:00",
"method": "POST",
"protocol": "HTTP/1.1",
"path": "/api/sample",
"query": {},
"headers": { "Content-Type": "application/json; charset=utf-8", "...": "..." },
"body_text": "{\"user_id\": 123, ...}",
"body_json": { "user_id": 123, "name": "太郎", "action": "sample" }
}
]
}履歴をクリアするには:
Invoke-RestMethod -Method Post http://localhost:8000/__clearservlet_client は送信時に 1 行ログを出すので、Lambda ログでも確認できます:
docker compose logs lambda[servlet_client] → POST http://mock-servlet:8000/api/sample (HTTP/1.1)
[servlet_client] ← status=200
さらに、モックサーブレットは受け取った内容をそのままエコーバックするため、
res.text を print すれば「自分が何を送ったか」を Lambda ログ内でも確認できます。
Lambda を通さずに、書式確認だけしたい場合:
# PowerShell の例 (GET)
Invoke-RestMethod "http://localhost:8000/api/test?id=1"
# PowerShell の例 (POST + JSON)
Invoke-RestMethod -Method Post -Uri http://localhost:8000/api/test `
-ContentType "application/json; charset=utf-8" `
-Body '{"hello": "world"}'ボディに日本語を含める場合は、直接起動の節と 同様に
[Text.Encoding]::UTF8.GetBytes(...)でバイト列にして渡してください (文字化け防止)。
lambda/servlet_client.py は標準ライブラリのみで動く小さな HTTP クライアントです
(pip install 不要)。handler.py から次のように使います。
from servlet_client import ServletClient
servlet = ServletClient() # 接続先は環境変数 SERVLET_BASE_URL で決まる| 書き方 | HTTP メソッド | 送信される書式 |
|---|---|---|
servlet.get("/api/x", params={"id": "1"}) |
GET | クエリ文字列 ?id=1 |
servlet.post_json("/api/x", {"a": 1}) |
POST | Content-Type: application/json の JSON ボディ |
servlet.post_form("/api/x", {"a": "1"}) |
POST | application/x-www-form-urlencoded (HTML フォームと同じ) |
servlet.put_json("/api/x/1", {"a": 2}) |
PUT | JSON ボディ |
servlet.delete("/api/x/1") |
DELETE | ボディなし |
servlet.request("PATCH", "/api/x", json_body={...}, headers={"X-Token": "abc"}) |
任意 | 任意 (低レベル API) |
戻り値は ServletResponse オブジェクト:
res = servlet.post_json("/api/sample", {"a": 1})
res.status # HTTP ステータスコード (int) 例: 200
res.text # 応答ボディ (str)
res.json() # 応答ボディを JSON として dict に変換
res.headers # 応答ヘッダー (dict)4xx / 5xx でも例外にはならず、res.status で判定できます。
servlet_client.pyを使わずurllibやhttp.clientを直接書いても構いません。 どんな書き方でも、実際に送られた内容はモックサーブレットが記録します。
「API キーを 暗号化された状態で SSM Parameter Store に置き、Lambda 側で復号して取り出す」 という処理を、AWS アカウント無しでそのまま検証できます。
mock-ssm コンテナは本物と同じ AWS の API プロトコルを話します。
そのため handler.py には 本物の AWS 向けと全く同じコード を書けます。
接続先の切り替えは docker-compose.yml の環境変数だけで済み、コード変更は不要です。
from ssm_client import SsmClient
ssm = SsmClient()
api_key = ssm.get_parameter("/lambda-verify/api-key") # 復号済みの平文が文字列で返るlambda/handler.py には既にこの例が入っているので、そのまま動かせます:
docker compose up -d
docker compose run --rm send
docker compose logs lambda[ssm_client] → GetParameter {"Name": "/lambda-verify/api-key", "WithDecryption": true} (http://mock-ssm:8082)
[ssm_client] ← OK (GetParameter)
[ssm] /lambda-verify/api-key = sk-t****************cdef (復号済み)
ssm_client.mask(値)を使うと、上のようにログで値を伏せられます。 平文をそのまま確認したいときは http://localhost:8082/__parameters を開いてください。
import boto3
ssm = boto3.client("ssm") # endpoint_url の指定は不要
res = ssm.get_parameter(Name="/lambda-verify/api-key", WithDecryption=True)
api_key = res["Parameter"]["Value"]docker-compose.yml の lambda サービスに AWS_ENDPOINT_URL_SSM: http://mock-ssm:8082
を設定してあります。これは boto3 が公式に見る環境変数なので、
boto3.client("ssm") と書くだけで自動的にモックへ接続されます
(boto3 は AWS 公式 Lambda イメージに標準で入っているため pip install も不要)。
本物の AWS に向けるときは、docker-compose.yml からこの 1 行を消すだけです。
handler.py のコードは 1 文字も変えなくて済みます。
戻り値の形・例外 (ssm.exceptions.ParameterNotFound など) も本物と同じです。
{
"/lambda-verify/api-key": {
"type": "SecureString",
"value": "sk-test-1234567890abcdef"
},
"/lambda-verify/feature-flag": "on"
}| 書き方 | 意味 |
|---|---|
"名前": "値" |
型は String (いちばん簡単な書き方) |
"名前": {"type": "SecureString", "value": "値"} |
暗号化パラメータ |
"名前": {"type": "StringList", "value": ["a", "b"]} |
カンマ区切りの文字列 a,b として返る |
"__メモ": "..." |
__ で始まるキーはコメント扱いで無視される |
- 保存するだけで反映されます (restart 不要)。 ファイルの更新を検知して自動で読み直します
- 先頭の
/は省略できます ("api-key"と"/api-key"は同じ扱い) - 反映結果は http://localhost:8082/__parameters ですぐ確認できます
個人用の値を使いたい場合は mock-ssm/parameters.local.json を作ると
parameters.json を上書きできます (.gitignore 済み)。
雛形は mock-ssm/parameters.local.json.example をコピーしてください。
Copy-Item mock-ssm/parameters.local.json.example mock-ssm/parameters.local.json本物と同じく、復号するかどうかで返る値が変わります。
ssm.get_parameter("/lambda-verify/api-key") # → sk-test-1234567890abcdef
ssm.get_parameter("/lambda-verify/api-key", with_decryption=False) # → AQICAHhc2stdGVzdC0xMjM0...ssm_clientのwith_decryptionは 既定True(すぐ平文が欲しい場面が多いため)- 素の boto3 の
WithDecryptionは本物と同じく 既定False - 暗号文は KMS 風の見た目にした疑似的なもの (base64) です。 「復号の有無で処理を分岐するコード」の検証には十分ですが、暗号強度はありません
from ssm_client import SsmClient, ParameterNotFound
import ssm_client
ssm = SsmClient()| 書き方 | 動作 |
|---|---|
ssm.get_parameter("/名前") |
値を文字列で返す (復号済み) |
ssm.get_parameter("/名前", with_decryption=False) |
暗号文のまま返す |
ssm.get_parameter("/名前", default="既定値") |
未登録なら例外ではなく既定値を返す |
ssm.get_parameter_detail("/名前") |
Value Type Version ARN を含む dict を返す |
ssm.get_parameters(["/a", "/b"]) |
まとめて取得し {名前: 値} の dict を返す |
ssm.get_parameters_by_path("/lambda-verify") |
パス配下をまとめて {名前: 値} で返す |
ssm.put_parameter("/名前", "値", type="SecureString") |
実行中に追加・更新 (メモリ上のみ。ファイルには書き戻さない) |
ssm_client.mask("値") |
ログ表示用に値を伏せる (sk-t****cdef) |
- 未登録の名前を
default無しで取るとParameterNotFoundが送出されます - 接続先は
AWS_ENDPOINT_URL_SSM→SSM_ENDPOINT_URL→AWS_ENDPOINT_URLの順に 解決されます。env_config経由で読むので、test.envや event の__test_envによる差し込みも効きます
# 登録されているパラメータ一覧 (平文・暗号文の両方が見える)
Invoke-RestMethod http://localhost:8082/__parameters | ConvertTo-Json -Depth 5
# handler がどのパラメータをどう取りに来たかの履歴
Invoke-RestMethod http://localhost:8082/__history | ConvertTo-Json -Depth 5
# モックのログ (呼び出しのたびに出力される)
docker compose logs -f mock-ssm------------------------------------------------------------
SSM 呼び出し #1 (2026-07-28T00:27:58)
API : GetParameter
対象 : /lambda-verify/api-key
復号 : する (WithDecryption=True) → 平文を返却
返した値 : 1 件 (/lambda-verify/api-key)
値 : sk-t****************cdef (平文は GET /__parameters で確認できます)
------------------------------------------------------------
| 管理用エンドポイント | 用途 |
|---|---|
GET http://localhost:8082/__parameters |
現在のパラメータ一覧 (平文・暗号文の両方) |
GET http://localhost:8082/__history |
SSM 呼び出しの履歴 (直近 100 件) |
POST http://localhost:8082/__reload |
定義ファイルを強制的に読み直す |
POST http://localhost:8082/__clear |
履歴をクリア |
対応している API は GetParameter / GetParameters / GetParametersByPath /
PutParameter / DescribeParameters です。
本物の Lambda では、環境変数を次のように読みます。
import os
base_url = os.environ.get("SERVLET_BASE_URL", "http://...") # 任意 (既定値あり)
token = os.environ["API_TOKEN"] # 必須 (無いとエラー)このローカル環境でも同じことはできますが、値を変えるたびに
docker-compose.yml を書き換えて起動し直すのは面倒です。
そこで 環境変数の取得を env_config.get_env(...) に置き換えるだけで、
docker-compose.yml を触らずにテスト値を「差し込める」 ようにしてあります。
# ── before: 素の os.environ を読む ──
import os
base_url = os.environ.get("SERVLET_BASE_URL", "http://mock-servlet:8000")
flag = os.environ.get("FEATURE_FLAG", "off")
# ── after: env_config 経由で読む (差し込みが効くようになる) ──
import env_config
base_url = env_config.get_env("SERVLET_BASE_URL", "http://mock-servlet:8000")
flag = env_config.get_env("FEATURE_FLAG", "off")引数の意味は os.environ.get(名前, 既定値) と全く同じです。
違いは、値の探し方が 「差し込み → 実際の環境変数 → 既定値」の順 になることだけです。
方法A: test.env ファイル |
方法B: event の __test_env |
|
|---|---|---|
| 範囲 | 全リクエスト共通 | その 1 回の起動だけ |
| 持続性 | 恒久 (消すまで有効) | 一時 (毎回 event で指定) |
| 反映方法 | docker compose restart lambda が必要 |
restart 不要 (送るたびに反映) |
| 主な用途 | 「このテスト期間中はずっとこの値」 | 「今回だけ値を変えて挙動を比べる」 |
SQS 送信経由 (send) |
✅ 効く | ❌ 効かない (※後述) |
直接起動 (:9000) |
✅ 効く | ✅ 効く |
優先順位 (強い順):
1. event の __test_env (方法B / 一時)
2. lambda/test.env (方法A / 恒久)
3. 実際の os.environ (docker-compose.yml で設定した値)
4. get_env(名前, 既定値) の既定値
上にあるものほど優先されます。たとえば docker-compose.yml に
FEATURE_FLAG=off があっても、test.env に FEATURE_FLAG=on と書けば
on が返り、さらにその起動の event に __test_env: {"FEATURE_FLAG": "debug"}
があれば debug が返ります。
1. サンプルをコピーして lambda/test.env を作る
# PowerShell
Copy-Item lambda/test.env.example lambda/test.env# Mac / Linux
cp lambda/test.env.example lambda/test.env2. lambda/test.env に KEY=VALUE 形式で書く
# 「#」で始まる行と空行は無視される
SERVLET_BASE_URL=http://host.docker.internal:8080
FEATURE_FLAG=on
API_TOKEN=test-token-12345
MAX_RETRY=3書式ルール:
- 1 行 1 つ、
KEY=VALUE #で始まる行・空行は無視- 値の前後の空白は自動で除去される。前後に空白を残したい場合は
"..."か'...'で囲む (例:NAME=" 太郎 ") - 先頭に
exportが付いていても読めます (export KEY=VALUE)
3. 反映する (restart が必要)
docker compose restart lambdaなぜ restart?
handler.pyの差し替えと同じ理由です。test.envは Lambda 起動時に一度だけ読み込まれ、ランタイムがメモリに保持するため、 変更後は restart で読み直しが必要です (数秒で完了)。
4. メッセージを送って確認
docker compose run --rm send
docker compose logs lambdalambda/test.env は .gitignore 済み なので、書いたテスト値が誤って
コミットされる心配はありません。チームで共有したいテンプレートは
lambda/test.env.example を編集してください。
送信する event のトップレベルに "__test_env" オブジェクトを入れると、
その起動の間だけ環境変数として差し込まれます。restart は不要で、送るたびに反映されます。
{
"__test_env": {
"FEATURE_FLAG": "on",
"SERVLET_BASE_URL": "http://mock-servlet:8000",
"API_TOKEN": "test-token-12345"
},
"Records": [
{ "body": "{\"user_id\": 123}", "...": "..." }
]
}- 値は自動的に文字列に変換されます (本物の環境変数は常に文字列のため。
3と書いてもget_envは"3"を返します) __test_envが無ければ何も差し込まれません (安全)handler.pyの先頭でenv_config.apply_event_overrides(event)を 1 回呼ぶ ことで有効になります (同梱のhandler.pyには既に入っています)
すぐ試せるサンプルとして sample-event-with-testenv.json を同梱しています。
直接起動 (次の節) でそのまま使えます:
# curl (Git Bash / Mac / Linux)
curl -X POST http://localhost:9000/2015-03-31/functions/function/invocations \
-d @sample-event-with-testenv.json# PowerShell (UTF-8 バイト列で送る)
$body = [Text.Encoding]::UTF8.GetBytes((Get-Content sample-event-with-testenv.json -Raw -Encoding UTF8))
Invoke-RestMethod -Method Post `
-Uri http://localhost:9000/2015-03-31/functions/function/invocations `
-ContentType "application/json; charset=utf-8" `
-Body $bodyLambda ログに次のように差し込み内容が表示されます:
[env_config] event から差し込まれた環境変数: {'FEATURE_FLAG': 'on', ...}
[env_config] SERVLET_BASE_URL=http://mock-servlet:8000 / FEATURE_FLAG=on
※
send(SQS 経由) では方法B は効きません。docker compose run --rm sendで送れるのは「メッセージ本文 (文字列)」だけで、 event 全体を組み立てるのは trigger コンテナです。event トップレベルの__test_envを付けられるのは、event を丸ごと自分で書ける直接起動のときだけです。 SQS 送信のフローで環境変数を差し込みたい場合は 方法A (test.env) を使ってください。
同梱の handler.py には既に組み込み済みですが、自分のスクリプトに書くときは
次の 3 ステップだけです。
import env_config
def lambda_handler(event, context):
# ① 方法B を有効化 (方法A だけ使う場合でも、呼んでおいて害はない)
env_config.apply_event_overrides(event)
# ② os.environ.get(...) の代わりに get_env(...) で読む
base_url = env_config.get_env("SERVLET_BASE_URL", "http://mock-servlet:8000")
flag = env_config.get_env("FEATURE_FLAG", "off")
# ③ 必須の環境変数は require_env(...) を使うと未設定時に分かりやすく落ちる
token = env_config.require_env("API_TOKEN") # 無ければ KeyError利用できる関数:
| 関数 | 相当する標準の書き方 | 動作 |
|---|---|---|
env_config.get_env(name, default=None) |
os.environ.get(name, default) |
優先順位付きで値を返す。無ければ default |
env_config.require_env(name) |
os.environ[name] |
優先順位付きで値を返す。無ければ KeyError |
env_config.apply_event_overrides(event) |
— | event の __test_env を一時差し込みとして有効化 (先頭で 1 回呼ぶ) |
env_config.active_env() |
— | 今効いている差し込み値 (方法A+方法B の実効値) を dict で返す (デバッグ表示用) |
env_config.clear_event_overrides() |
— | 一時差し込み (方法B) を手動でクリア (通常は不要) |
SERVLET_BASE_URLについて:servlet_client.pyも内部でenv_config経由でSERVLET_BASE_URLを読むようにしてあります。そのため方法A / 方法B で差し込んだ URL は、ServletClientの接続先にもそのまま反映されます (リクエストのたびに 解決するので、方法B のように起動ごとに変わる値でも正しく効きます)。
キューを経由せずに、任意のイベントで handler.py を即座に実行できます。
イベントは sample-event.json (本物と同じ SQS イベント形式のサンプル) を使います。
# PowerShell (日本語が化けないよう UTF-8 のバイト列で送る)
$body = [Text.Encoding]::UTF8.GetBytes((Get-Content sample-event.json -Raw -Encoding UTF8))
Invoke-RestMethod -Method Post `
-Uri http://localhost:9000/2015-03-31/functions/function/invocations `
-ContentType "application/json; charset=utf-8" `
-Body $body注意:
-Bodyに文字列をそのまま渡すと、Windows PowerShell 5.1 では日本語が 文字化けします (ISO-8859-1 でエンコードされるため)。必ず上記のようにバイト列で渡してください。
# curl (Git Bash / Mac / Linux)
curl -X POST http://localhost:9000/2015-03-31/functions/function/invocations \
-d @sample-event.jsonsample-event.json の body フィールドを書き換えれば、任意のメッセージ本文で試せます
(body は「JSON を文字列化したもの」である点に注意)。
lambda_handler(event, context) の event には、本物の AWS と同じ以下の形式が渡されます:
{
"Records": [
{
"messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
"receiptHandle": "AQEB...",
"body": "ここに SQS メッセージ本文 (文字列)",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1753142400000",
"SenderId": "000000000000",
"ApproximateFirstReceiveTimestamp": "1753142400001"
},
"messageAttributes": {},
"md5OfBody": "...",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-1:000000000000:lambda-queue",
"awsRegion": "us-east-1"
}
]
}ポイント:
- メッセージ本文は
event["Records"][i]["body"]に 文字列 で入る (JSON を送った場合も文字列なのでjson.loads()が必要) - 複数メッセージが 1 回の起動にまとめられることがある (最大 10 件) ため、
必ず
Recordsをループで処理する attributes["ApproximateReceiveCount"]で「何回目の配信か」(再試行回数) が分かる
書式の確認が済んだら、モックではなく実物のサーブレットに向けることもできます。
docker-compose.yml の lambda サービスの環境変数を書き換えるだけです:
lambda:
environment:
# ホスト PC (Windows) 上で動いている Tomcat 等に向ける場合:
SERVLET_BASE_URL: http://host.docker.internal:8080
# 社内サーバー等に向ける場合:
# SERVLET_BASE_URL: http://servlet.example.co.jp:8080変更後は反映のため:
docker compose up -d lambda
host.docker.internalは「コンテナから見たホスト PC」を指す Docker Desktop の特殊ホスト名です。 ホストで Tomcat をポート 8080 で動かしていればそのまま届きます。
handler.py 側のコードは一切変更不要です (ServletClient が環境変数を読むため)。
handler.py で requests や pandas などを使いたい場合は、Lambda イメージを 1 枚かぶせます。
-
lambda/Dockerfileを作成:FROM public.ecr.aws/lambda/python:3.13 RUN pip install --no-cache-dir requests pandas
-
docker-compose.ymlのlambdaサービスをimage:からbuild:に変更:lambda: build: ./lambda command: ["handler.lambda_handler"] # 以下は元のまま
-
再ビルドして起動:
docker compose up -d --build lambda
補足:
boto3は AWS 公式イメージに最初から入っているため追加不要です。
| やりたいこと | コマンド |
|---|---|
| 環境を起動 | docker compose up -d --build |
| 環境を停止・削除 | docker compose down |
| handler.py の変更を反映 | docker compose restart lambda |
| 環境変数を差し込む (方法A) | lambda/test.env に KEY=VALUE を書いて docker compose restart lambda |
| 環境変数を差し込む (方法B) | 直接起動する event に "__test_env": {...} を含める (restart 不要) |
| SSM パラメータを追加・変更 | mock-ssm/parameters.json を編集して保存するだけ (restart 不要) |
| テストメッセージ送信 (message.json) | docker compose run --rm send |
| テストメッセージ送信 (文字列) | docker compose run --rm send "任意の文字列" |
| Lambda のログを見る | docker compose logs lambda |
| サーブレット受信内容を見る | docker compose logs mock-servlet |
| SSM 呼び出し内容を見る | docker compose logs mock-ssm |
| 全ログをリアルタイム監視 | docker compose logs -f |
| コンテナの状態確認 | docker compose ps |
| サーブレット受信履歴 (JSON) | ブラウザで http://localhost:8000/__history |
| サーブレット受信履歴クリア | Invoke-RestMethod -Method Post http://localhost:8000/__clear |
| SSM パラメータ一覧 (JSON) | ブラウザで http://localhost:8082/__parameters |
| SSM 呼び出し履歴 (JSON) | ブラウザで http://localhost:8082/__history |
| キューの状態を見る | docker compose run --rm status |
Container_Compose_Lambda_Python/
├── docker-compose.yml # Compose 定義 (全コンテナの構成)
├── README.md # このファイル
├── .gitignore # lambda/test.env などをコミットから除外
├── message.json # ★ 送信するテストメッセージ (自由に編集)
├── sample-event.json # Lambda 直接起動用の SQS イベントサンプル
├── sample-event-with-testenv.json # 環境変数差し込み (方法B) 付きの直接起動サンプル
├── elasticmq/
│ └── elasticmq.conf # キュー定義 (lambda-queue が自動作成される)
├── lambda/
│ ├── handler.py # ★★ 検証したいスクリプトはここだけ差し替える ★★
│ ├── servlet_client.py # サーブレット呼び出し用クライアント (編集不要)
│ ├── ssm_client.py # Parameter Store 取得用クライアント (編集不要)
│ ├── env_config.py # 環境変数のテスト差し込みヘルパー (編集不要)
│ ├── test.env.example # ★ 方法A のサンプル。test.env にコピーして使う
│ └── test.env # ★ 方法A の環境変数 (作成すると使われる。.gitignore 済み)
├── mock-servlet/
│ └── server.py # モックサーブレット本体 (編集不要)
├── mock-ssm/
│ ├── server.py # モック SSM Parameter Store 本体 (編集不要)
│ ├── parameters.json # ★ 取得できるパラメータの定義 (自由に編集)
│ ├── parameters.local.json.example # 個人用上書きのサンプル
│ └── parameters.local.json # ★ 個人用の上書き (作成すると使われる。.gitignore 済み)
└── trigger/
├── Dockerfile # trigger / send / status 用イメージ
├── poller.py # SQS→Lambda トリガー (編集不要)
├── send.py # メッセージ送信ツール (編集不要)
└── status.py # キュー状態確認ツール (編集不要)
★ 印のファイルだけ触れば日常の検証は完結します。
docker compose psで 5 コンテナすべてがUpか確認docker compose logs triggerを確認- 「キュー待機中...」を繰り返す → elasticmq が起動していない。
docker compose logs elasticmqを確認 - 「Lambda へ接続できません」→ lambda コンテナが落ちている。
docker compose logs lambdaを確認
- 「キュー待機中...」を繰り返す → elasticmq が起動していない。
docker compose restart lambda を実行しましたか? 反映には restart が必要です。
docker compose logs lambda に Python のエラー (SyntaxError 等) が出ます。
修正して docker compose restart lambda してください。
エラーで失敗したメッセージはキューに残っており、約 30 秒ごとに自動で再実行されるため、
修正して restart するだけで再送は不要です (再実行をやめたい場合は
docker compose restart elasticmq でキューを空にできます — ElasticMQ はメモリ保持のため)。
- ファイル名は
lambda/test.envになっていますか? (test.env.exampleのままだと読まれません) docker compose restart lambdaを実行しましたか? 方法A (test.env) は反映に restart が必要です。handler.pyでos.environ.get(...)のままになっていませんか? 差し込みを効かせるにはenv_config.get_env(...)で読む必要があります。docker-compose.ymlに同名の環境変数があっても問題ありません (優先順位はtest.envの方が上です)。それでも変わらない場合はdocker compose logs lambdaの[env_config] ...行で実効値を確認してください。
send(SQS 経由) では方法B は効きません。event 全体を書ける直接起動で使ってください (詳細は環境変数をテスト用に差し込むの注記)。__test_envは event のトップレベルに置きます (Recordsの中ではありません)。handler.pyの先頭でenv_config.apply_event_overrides(event)を呼んでいますか?
- 名前は
mock-ssm/parameters.jsonに登録済みですか? http://localhost:8082/__parameters で現在の登録内容を確認できます parameters.jsonが JSON として壊れていませんか?docker compose logs mock-ssmに[エラー] ... JSON として読めませんが出ます (このとき直前に読めていた内容で動き続けます)mock-ssmコンテナはUpですか? (docker compose ps) 落ちているとConnection refusedやEndpointConnectionErrorになります
復号指定が漏れています。素の boto3 は WithDecryption の既定が False のため、
ssm.get_parameter(Name="...", WithDecryption=True) と明示してください
(ssm_client.get_parameter() は既定で復号します)。
handler がエラー (例外) を返し続けているためです。本物の SQS トリガーと同じ挙動です。
docker compose logs lambda でエラー原因を確認・修正してください。
止めたい場合は docker compose restart elasticmq でキューをリセットします。
9324 9000 8000 8082 のどれかが使用中です。docker-compose.yml の
ports: の左側 (ホスト側) の番号を空いている番号に変えてください
(例: "18000:8000" にすると履歴 URL は http://localhost:18000/__history になる)。
ホスト側の番号を変えても、コンテナ同士の通信 (
mock-servlet:8000/mock-ssm:8082) には影響しないのでhandler.pyの修正は不要です。
PowerShell の curl は Invoke-WebRequest の別名で、Linux の curl と挙動が違います。
本 README の PowerShell 例 (Invoke-RestMethod) を使うか、curl.exe と明示してください。
message.json や handler.py は UTF-8 (BOM なし) で保存してください。
VS Code なら右下のエンコーディング表示から変更できます。
docker compose down
docker compose up -d --build --force-recreateこのローカル環境は検証用の簡易再現であり、以下は本物と異なります:
| 項目 | 本物の AWS | この環境 |
|---|---|---|
| キュー | Amazon SQS | ElasticMQ (SQS 互換 API)。メッセージはメモリ保持のため、コンテナを止めると消える |
| トリガー | イベントソースマッピング | trigger コンテナによるポーリング再現 |
| 失敗時の再試行 | 可視性タイムアウト後に再配信 (同じ) | 同じ。ただし DLQ (デッドレターキュー) や maxReceiveCount は未設定のため、失敗し続けると無限に再試行される |
部分バッチ応答 (batchItemFailures) |
対応 | 未対応 (バッチ全体で成功/失敗を判定) |
| Lambda タイムアウト | 設定値で強制終了 | 実質無制限 (RIE の仕様) |
| 同時実行・スケーリング | 自動スケール | 常に 1 並列 |
| IAM 権限 | 実際に検証される | ダミー認証情報のため権限エラーは再現されない |
| Parameter Store | AWS Systems Manager | mock-ssm コンテナ。SecureString の暗号化は KMS ではなく見た目だけの疑似暗号 (復号の有無による分岐は再現できる)。バージョン履歴・ラベル・パラメータポリシー・ページネーションは未対応 |
「SQS メッセージ受信 → Python スクリプトの処理 → サーブレット呼び出しの書式」の検証という 目的においては、本物と同じイベント形式・同じ Python 3.13 ランタイムで確認できます。