AWS Lambda Python入門｜3.14対応handler例とローカルテスト

AWS Lambda の Python 関数とは、lambda_handler(event, context) という形の関数を1つ書き、「ファイル名.関数名」をハンドラとして登録するだけで、API・S3・SQS・スケジュールなどのイベントに応じて実行されるサーバーレスのプログラムです。 2026年10月時点の公式ランタイム一覧では python3.14 が利用でき（python3.15 も掲載済み）、この記事のコードはすべて Python 3.14.7 で動かして確認しています。

この記事では、handler の書き方と event/context の中身、トリガー別（API Gateway HTTP API v2・S3・SQS・EventBridge）のコピペできる event JSON と handler 例、環境変数・boto3・ログ・タイムアウト・コールドスタートの考え方、pytest と moto（mock_aws）によるローカルテストの実行結果、zip パッケージ化とデプロイ、そして公式ランタイムクライアント（awslambdaric 4.1.0）で実際に再現したエラー文の対処表までを一気にまとめます。

Lambda Python の早見表

lambda_handler の書き方：名前・event・context・戻り値

handler の名前は「ファイル名.関数名」

Lambda は「ハンドラ」設定に書かれた モジュール名.関数名 を import して呼び出します。 ファイルが lambda_function.py、関数が lambda_handler なら設定値は lambda_function.lambda_handler。ファイル名を app.py にしたら app.lambda_handler と、設定も合わせて変えます。関数名そのものは固定ではありません。

event と context の中身

戻り値のルール

戻り値は Lambda ランタイムが json.dumps 相当で文字列化して返します。 datetime・Decimal・set をそのまま返すと Runtime.MarshalError（後述）になるので、isoformat() や float() で変換してから返します。

トリガー別の event サンプルと handler（コピペ用）

トリガーごとに event の形も「期待される戻り値」も違うので、まず表で全体を押さえます。

API Gateway HTTP API（v2）：POST /hello を受ける

events/apigw_http_v2.json（ペイロード形式 2.0）:

{
  "version": "2.0",
  "routeKey": "POST /hello",
  "rawPath": "/hello",
  "rawQueryString": "",
  "headers": {"content-type": "application/json", "host": "abc123.execute-api.ap-northeast-1.amazonaws.com", "user-agent": "curl/8.5.0"},
  "requestContext": {
    "accountId": "123456789012", "apiId": "abc123",
    "domainName": "abc123.execute-api.ap-northeast-1.amazonaws.com",
    "http": {"method": "POST", "path": "/hello", "protocol": "HTTP/1.1", "sourceIp": "203.0.113.10", "userAgent": "curl/8.5.0"},
    "requestId": "JKJaXmPLvHcESHA=", "routeKey": "POST /hello", "stage": "$default",
    "time": "01/Oct/2026:09:00:00 +0000", "timeEpoch": 1790845200000
  },
  "body": "{\"name\": \"Lambda\"}",
  "isBase64Encoded": false
}

src/lambda_function.py（ハンドラ設定: lambda_function.lambda_handler）:

import json
import logging
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)  # Lambda の既定は WARNING。INFO を出したいときに指定する

STAGE = os.environ.get("STAGE", "dev")  # 環境変数はハンドラの外で1回だけ読む


def lambda_handler(event, context):
    http = event["requestContext"]["http"]
    logger.info("request_id=%s %s %s", context.aws_request_id, http["method"], event["rawPath"])

    try:
        body = json.loads(event["body"]) if event.get("body") else {}
    except json.JSONDecodeError:
        return _response(400, {"error": "body must be JSON"})

    query = event.get("queryStringParameters") or {}
    name = body.get("name") or query.get("name") or "world"

    return _response(200, {"message": f"Hello, {name}!", "stage": STAGE,
                           "remaining_ms": context.get_remaining_time_in_millis()})


def _response(status, payload):
    """HTTP API v2 が期待する戻り値の形式に整える"""
    return {
        "statusCode": status,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(payload, ensure_ascii=False),
    }

公式ランタイムクライアント（awslambdaric 4.1.0、Python 3.14.7）にこの event を渡したときの応答です（環境変数 STAGE=prod）:

{
  "statusCode": 200,
  "headers": {"Content-Type": "application/json"},
  "body": "{\"message\": \"Hello, Lambda!\", \"stage\": \"prod\", \"remaining_ms\": 2999}"
}

body は文字列である点が最大のポイントです。 受け取るときは json.loads、返すときは json.dumps を通します。queryStringParameters はクエリが無いとキー自体が無い（None）ので or {} で受けます。

S3：ファイルが置かれたら集計して書き戻す

events/s3_put.json:

{
  "Records": [
    {
      "eventVersion": "2.1", "eventSource": "aws:s3", "awsRegion": "ap-northeast-1",
      "eventTime": "2026-10-01T09:00:00.000Z", "eventName": "ObjectCreated:Put",
      "s3": {
        "s3SchemaVersion": "1.0", "configurationId": "orders-upload",
        "bucket": {"name": "sample-orders-bucket", "ownerIdentity": {"principalId": "A3EXAMPLE"}, "arn": "arn:aws:s3:::sample-orders-bucket"},
        "object": {"key": "uploads/orders+2026-10-01.csv", "size": 25, "eTag": "0123456789abcdef0123456789abcdef", "sequencer": "0A1B2C3D4E5F678901"}
      }
    }
  ]
}

src/s3_handler.py（ハンドラ設定: s3_handler.lambda_handler）:

import json
import logging
import os
import urllib.parse

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3 = boto3.client("s3")  # ハンドラの外で生成 → ウォームスタート時に再利用される
SUMMARY_PREFIX = os.environ.get("SUMMARY_PREFIX", "summary/")


def lambda_handler(event, context):
    results = []
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        # S3 イベントのキーは URL エンコード済み（空白は "+"）なので必ず戻す
        key = urllib.parse.unquote_plus(record["s3"]["object"]["key"])

        obj = s3.get_object(Bucket=bucket, Key=key)
        text = obj["Body"].read().decode("utf-8")
        rows = [line for line in text.splitlines() if line.strip()]

        summary = {"source": key, "rows": len(rows) - 1, "size": obj["ContentLength"]}
        summary_key = f"{SUMMARY_PREFIX}{key}.json"
        s3.put_object(Bucket=bucket, Key=summary_key,
                      Body=json.dumps(summary, ensure_ascii=False),
                      ContentType="application/json")
        logger.info("summarized %s -> %s", key, summary_key)
        results.append(summary)

    return {"processed": len(results), "results": results}

キーに空白や日本語が含まれると orders+2026-10-01.csv のようにエンコードされて届きます。 unquote_plus を通さずに get_object すると NoSuchKey になるので、S3 トリガーでは定型として入れておきます。実行結果は後述の moto テストで確認します。

SQS：失敗したメッセージだけ再配信させる

events/sqs.json（2件目はわざと JSON でない本文）:

{
  "Records": [
    {
      "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
      "receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
      "body": "{\"orderId\": \"A-1001\", \"qty\": 2}",
      "attributes": {"ApproximateReceiveCount": "1"}, "messageAttributes": {},
      "eventSource": "aws:sqs", "eventSourceARN": "arn:aws:sqs:ap-northeast-1:123456789012:order-queue", "awsRegion": "ap-northeast-1"
    },
    {
      "messageId": "2e1424d4-f796-459a-8184-9c92662be6da",
      "receiptHandle": "AQEBzWwaftRI0KuVm4tP+/7q1rGgNqicHq...",
      "body": "this is not json",
      "attributes": {"ApproximateReceiveCount": "1"}, "messageAttributes": {},
      "eventSource": "aws:sqs", "eventSourceARN": "arn:aws:sqs:ap-northeast-1:123456789012:order-queue", "awsRegion": "ap-northeast-1"
    }
  ]
}

src/sqs_handler.py（ハンドラ設定: sqs_handler.lambda_handler）:

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
    failures = []
    for record in event["Records"]:
        try:
            order = json.loads(record["body"])
            logger.info("order %s qty=%s", order["orderId"], order["qty"])
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning("skip messageId=%s reason=%r", record["messageId"], e)
            failures.append({"itemIdentifier": record["messageId"]})

    # 失敗した messageId だけ返すと、その分だけ再配信される（成功分は削除される）
    return {"batchItemFailures": failures}

この戻り値を有効にするには、イベントソースマッピングで「部分バッチ応答（ReportBatchItemFailures）」を有効にします。 有効にしないまま例外を投げると、成功した1件目まで含めてバッチ全体がキューに戻ります。SQS の可視性タイムアウトは関数タイムアウトの 6 倍以上、というのが公式の推奨です。

EventBridge スケジュール：毎晩の掃除処理

events/eventbridge_schedule.json（EventBridge ルールの rate() / cron() が送る形）:

{
  "version": "0",
  "id": "53dc4d37-cffa-4f76-80c9-8b7d4a4d2eaa",
  "detail-type": "Scheduled Event",
  "source": "aws.events",
  "account": "123456789012",
  "time": "2026-10-01T09:00:00Z",
  "region": "ap-northeast-1",
  "resources": ["arn:aws:events:ap-northeast-1:123456789012:rule/nightly-cleanup"],
  "detail": {}
}

src/schedule_handler.py（ハンドラ設定: schedule_handler.lambda_handler）:

import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
    fired_at = datetime.fromisoformat(event["time"].replace("Z", "+00:00"))
    logger.info("scheduled run: %s (rule=%s)", fired_at.isoformat(), event["resources"][0])

    cleaned = 0
    # 残り時間を見ながら処理する（長いバッチはここで打ち切って次回に回す）
    while context.get_remaining_time_in_millis() > 1000 and cleaned < 3:
        cleaned += 1  # 実際には1件ずつ掃除する処理

    return {"fired_at": fired_at.astimezone(timezone.utc).isoformat(), "cleaned": cleaned}

新しい EventBridge Scheduler を使う場合、event はスケジュール作成時に指定した入力 JSON（既定は空）になります。 どちらでも動かしたいなら、時刻は event から取らず datetime.now(timezone.utc) で取ると決めておきます。

環境変数・boto3・ログ・タイムアウト・コールドスタート

環境変数は handler の外で読む

環境変数はコンソールや --environment で設定し、os.environ で読みます。 モジュール直下で読めば実行環境の初期化時に1回だけ評価され、ウォームスタートでは再利用されます。Lambda が自動で入れる AWS_REGION、AWS_LAMBDA_FUNCTION_NAME、AWS_LAMBDA_FUNCTION_MEMORY_SIZE などの予約変数も同じ方法で読めます。

boto3 はランタイム同梱、クライアントは handler の外で生成

boto3 は Python ランタイムに同梱されているので、import するだけで使えます。 同梱版のバージョンは Lambda 側の更新で変わるため、特定バージョンに依存する処理は自分のパッケージに同梱します。boto3.client("s3") の生成は認証情報の解決などで時間がかかるので handler の外に出して再利用します。権限は実行ロールに付けます（ログ用の AWSLambdaBasicExecutionRole に加え、S3 なら s3:GetObject/s3:PutObject）。

ログ：logging を使うと CloudWatch にこう出る

print も logging も CloudWatch Logs に届きますが、logging を使うとレベル・時刻・リクエスト ID が自動で付きます。 公式ランタイムクライアントで次の関数を動かした出力です。

import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
    print("print で出した行")
    logger.info("logger.info で出した行 %s", {"items": 3})
    logger.warning("logger.warning で出した行")
    return {"ok": True}

テキスト形式（既定）:

print で出した行
[INFO]  2026-10-01T03:38:40.548Z    11111111-2222-3333-4444-555555555555    logger.info で出した行 {'items': 3}
[WARNING]   2026-10-01T03:38:40.548Z    11111111-2222-3333-4444-555555555555    logger.warning で出した行

JSON 形式（関数のログ設定で JSON を選ぶ。環境変数 AWS_LAMBDA_LOG_FORMAT=JSON・AWS_LAMBDA_LOG_LEVEL=INFO に相当）:

print で出した行
{"timestamp": "2026-10-01T03:38:41Z", "level": "INFO", "message": "logger.info で出した行 {'items': 3}", "logger": "root", "requestId": "11111111-2222-3333-4444-555555555555"}
{"timestamp": "2026-10-01T03:38:41Z", "level": "WARNING", "message": "logger.warning で出した行", "logger": "root", "requestId": "11111111-2222-3333-4444-555555555555"}

ルートロガーの既定レベルは WARNING なので、logger.setLevel(logging.INFO) を書かないと logger.info は出ません。 JSON 形式なら CloudWatch Logs Insights で filter level = "ERROR" のように絞れます。実際の Lambda では、これらの前後にサービスが付ける START RequestId: ...、END RequestId: ...、REPORT RequestId: ... Duration: ... Billed Duration: ... Memory Size: ... Max Memory Used: ... Init Duration: ... の行が並びます。

タイムアウトとメモリ

残り時間は context.get_remaining_time_in_millis() で取れるので、ループ処理では「残り 1 秒を切ったら打ち切って次回に回す」設計にすると、途中状態のまま Task timed out で終わることを防げます（上の schedule_handler.py がその形）。

コールドスタートの考え方

コールドスタートとは、新しい実行環境が作られるときに「ランタイム起動 → モジュール import → handler 外のコード実行」が走る時間のことです。 続けて呼ばれるときは同じ環境が再利用（ウォームスタート）され、この初期化は省略されます。

ローカルテスト：pytest で handler を直接呼ぶ

Lambda の handler はただの Python 関数なので、event（dict）とフェイクの context を渡せば pytest でそのままテストできます。 構成と環境は次のとおりです（Python 3.14.7、pytest 9.1.1、boto3 1.43.106、moto 5.2.3）。

lambda-python-sample/
├── pytest.ini                 # [pytest] pythonpath = src tests / testpaths = tests
├── src/                       # lambda_function.py, s3_handler.py, sqs_handler.py, schedule_handler.py
├── events/                    # 上で載せた 4 つの JSON
└── tests/
    ├── conftest.py            # FakeContext とダミー認証情報
    ├── test_handlers.py
    └── test_s3_handler.py     # moto で S3 を模擬

uv python install 3.14            # ランタイムとローカルの Python を揃える
uv venv --python 3.14 .venv && source .venv/bin/activate
uv pip install pytest boto3 "moto[s3]"

tests/conftest.py:

import json
import os
from pathlib import Path

import pytest

# boto3 が必要とするリージョンと認証情報をダミーで固定する（本物のキーが使われないようにする）
os.environ["AWS_DEFAULT_REGION"] = "ap-northeast-1"
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
os.environ["STAGE"] = "test"  # 関数が読む環境変数もここで用意する

EVENTS_DIR = Path(__file__).resolve().parent.parent / "events"


class FakeContext:
    """本番の context と同じ属性名をもつフェイク（必要な分だけ）"""

    function_name = "sample-function"
    invoked_function_arn = "arn:aws:lambda:ap-northeast-1:123456789012:function:sample-function"
    memory_limit_in_mb = 128
    aws_request_id = "11111111-2222-3333-4444-555555555555"

    def __init__(self, remaining_ms: int = 3000):
        self._remaining_ms = remaining_ms

    def get_remaining_time_in_millis(self) -> int:
        return self._remaining_ms


@pytest.fixture
def context():
    return FakeContext()


@pytest.fixture
def load_event():
    def _load(name: str) -> dict:
        return json.loads((EVENTS_DIR / name).read_text(encoding="utf-8"))

    return _load

tests/test_handlers.py（API Gateway・SQS・スケジュールを直接呼ぶ）:

import json

from lambda_function import lambda_handler as api_handler
from schedule_handler import lambda_handler as schedule_handler
from sqs_handler import lambda_handler as sqs_handler


def test_api_post_body(load_event, context):
    res = api_handler(load_event("apigw_http_v2.json"), context)
    assert res["statusCode"] == 200
    assert json.loads(res["body"]) == {"message": "Hello, Lambda!", "stage": "test", "remaining_ms": 3000}


def test_api_invalid_json_is_400(load_event, context):
    event = load_event("apigw_http_v2.json")
    event["body"] = "{broken"
    res = api_handler(event, context)
    assert res["statusCode"] == 400
    assert json.loads(res["body"]) == {"error": "body must be JSON"}


def test_sqs_reports_only_failed_message(load_event, context):
    res = sqs_handler(load_event("sqs.json"), context)
    assert res == {"batchItemFailures": [{"itemIdentifier": "2e1424d4-f796-459a-8184-9c92662be6da"}]}


def test_schedule_stops_when_time_is_short(load_event, context):
    event = load_event("eventbridge_schedule.json")
    assert schedule_handler(event, context)["cleaned"] == 3
    # 残り時間が 1 秒を切っている状況を再現する
    from conftest import FakeContext
    assert schedule_handler(event, FakeContext(remaining_ms=500))["cleaned"] == 0

moto の mock_aws で S3 を伴う処理をテストする

moto の @mock_aws を付けたテストの中では、boto3 の呼び出しがすべてメモリ上の偽 AWS に向かいます。 バケットを作り、ファイルを置き、S3 イベントで handler を呼び、書き戻された JSON まで検証します。

tests/test_s3_handler.py:

import json

import boto3
from moto import mock_aws  # s3_handler より先に import する（モジュール内の boto3.client を横取りさせるため）

import s3_handler

BUCKET = "sample-orders-bucket"
KEY = "uploads/orders 2026-10-01.csv"  # イベント側は "orders+2026-10-01.csv"（URLエンコード済み）
CSV = "id,qty\nA-1001,2\nA-1002,5\n"


@mock_aws
def test_s3_put_creates_summary(load_event, context):
    s3 = boto3.client("s3")
    s3.create_bucket(Bucket=BUCKET, CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"})
    s3.put_object(Bucket=BUCKET, Key=KEY, Body=CSV.encode("utf-8"))

    result = s3_handler.lambda_handler(load_event("s3_put.json"), context)

    assert result == {"processed": 1, "results": [{"source": KEY, "rows": 2, "size": 25}]}
    saved = s3.get_object(Bucket=BUCKET, Key=f"summary/{KEY}.json")
    assert json.loads(saved["Body"].read()) == result["results"][0]


@mock_aws
def test_s3_missing_object_raises(load_event, context):
    import pytest
    from botocore.exceptions import ClientError

    boto3.client("s3").create_bucket(Bucket=BUCKET,
                                     CreateBucketConfiguration={"LocationConstraint": "ap-northeast-1"})
    with pytest.raises(ClientError) as ei:
        s3_handler.lambda_handler(load_event("s3_put.json"), context)
    assert ei.value.response["Error"]["Code"] == "NoSuchKey"

実行結果（python -m pytest -v）:

============================= test session starts ==============================
platform linux -- Python 3.14.7, pytest-9.1.1, pluggy-1.6.0
configfile: pytest.ini
testpaths: tests
collecting ... collected 6 items

tests/test_handlers.py::test_api_post_body PASSED                        [ 16%]
tests/test_handlers.py::test_api_invalid_json_is_400 PASSED              [ 33%]
tests/test_handlers.py::test_sqs_reports_only_failed_message PASSED      [ 50%]
tests/test_handlers.py::test_schedule_stops_when_time_is_short PASSED    [ 66%]
tests/test_s3_handler.py::test_s3_put_creates_summary PASSED             [ 83%]
tests/test_s3_handler.py::test_s3_missing_object_raises PASSED           [100%]

============================== 6 passed in 0.53s ===============================

6 件すべて 0.53 秒で通り、S3 の + を空白に戻す処理、部分バッチ応答、残り時間での打ち切りまでローカルだけで確認できました。 なお moto で Lambda の invoke 自体（boto3.client("lambda").invoke）を模擬する機能は Docker が必要なため、この記事では実行していません。handler を直接呼ぶ方式なら Docker なしで済みます。

moto でつまずく「import の順番」

boto3.client(...) を handler の外で作っている関数は、moto を import するより前にそのモジュールを import すると、クライアントがモックされません。 実際に順番を入れ替えて試すと、list_buckets() が偽 AWS ではなく本物のエンドポイントへ接続しに行きました（この検証環境ではネットワーク設定により ProxyConnectionError で止まりましたが、本物の認証情報がある環境ではそのまま本物の S3 に届きます）。

パッケージ化とデプロイ

zip パッケージ：pip install --target と zip

ランタイムに入っていないライブラリ（例：httpx）は、pip install --target で関数と同じ階層に展開して zip にまとめます。 下の lambda_function.py を同梱なしで動かすと Unable to import module 'lambda_function': No module named 'httpx' になり、同梱すると動く、という流れを実際に確認しました。

# lambda_function.py
import json

import httpx  # サードパーティのライブラリ（デプロイパッケージに同梱が必要）


def lambda_handler(event, context):
    return {"statusCode": 200, "body": json.dumps({"httpx": httpx.__version__})}

python -m pip install --target ./package httpx      # 依存を package/ に展開
cd package && zip -r ../function.zip . && cd ..     # 依存を zip の「ルート」に入れる
zip function.zip lambda_function.py                 # 関数ファイルもルートに追加
unzip -l function.zip | grep -E "lambda_function.py|httpx/$"
#        0  2026-10-01 12:39   httpx/
#      239  2026-10-01 12:39   lambda_function.py

zip を展開した状態（Lambda の /var/task 相当）で公式ランタイムクライアントから呼んだ応答:

{"statusCode": 200, "body": "{\"httpx\": \"0.28.1\"}"}

C 拡張を含むライブラリ（pydantic、numpy など）は、手元が macOS や Windows でも Lambda 用（Linux x86_64、Python 3.14）のホイールを指定して取得します。

python -m pip install --target ./package \
  --platform manylinux2014_x86_64 --only-binary=:all: --python-version 3.14 pydantic
ls package/pydantic_core | grep "\.so$"
# _pydantic_core.cpython-314-x86_64-linux-gnu.so   ← Lambda(x86_64) で動くバイナリ

arm64（Graviton）で動かす関数なら --platform manylinux2014_aarch64 にします。zip は直接アップロードで 50 MB、展開後 250 MB（レイヤー含む）が上限で、超える場合はコンテナイメージ（最大 10 GB）を使います。

Lambda レイヤー：共通ライブラリを分けて持つ

レイヤーは python/ フォルダの下にライブラリを置いた zip です。 複数の関数で同じ依存を共有でき、関数本体の zip が小さくなります（1 関数あたり最大 5 層）。

python -m pip install --target ./layer/python httpx
cd layer && zip -r ../layer.zip python && cd ..
unzip -l layer.zip | sed -n 4p
#        0  2026-10-01 12:39   python/

AWS CLI でデプロイするコマンド（本物の AWS では未実行）

この記事の検証環境は AWS に接続できないため、以下は公式ドキュメントに沿った形で示します。 実行ロール（AWSLambdaBasicExecutionRole を付けたロール）は事前に作っておきます。

# 新規作成
aws lambda create-function \
  --function-name sample-function \
  --runtime python3.14 \
  --architectures x86_64 \
  --handler lambda_function.lambda_handler \
  --role arn:aws:iam::123456789012:role/lambda-basic-execution \
  --zip-file fileb://function.zip \
  --timeout 10 --memory-size 256 \
  --environment "Variables={STAGE=prod}"

# コードだけ更新
aws lambda update-function-code --function-name sample-function --zip-file fileb://function.zip

# レイヤーを公開して関数に付ける
aws lambda publish-layer-version --layer-name httpx-layer --zip-file fileb://layer.zip \
  --compatible-runtimes python3.14 --compatible-architectures x86_64
aws lambda update-function-configuration --function-name sample-function \
  --layers arn:aws:lambda:ap-northeast-1:123456789012:layer:httpx-layer:1

# 手元の event JSON で呼び出す
aws lambda invoke --function-name sample-function \
  --cli-binary-format raw-in-base64-out --payload file://events/apigw_http_v2.json out.json

SAM / CDK で IaC 化する（概要）

よくあるエラーと対処表

以下のエラー文は、公式ランタイムクライアント awslambdaric 4.1.0（Python 3.14.7）に実際のハンドラ設定・コードを渡して再現したものです（Task timed out とメモリ超過は Lambda サービス側が出すため、公式ドキュメントの形式で記載）。

例外が起きたときに Lambda が返すペイロードの形も押さえておくと、呼び出し側（API や Step Functions）での扱いが決めやすくなります。 上の KeyError の実物です（パスは一部省略。実際の Lambda では /var/task/lambda_function.py になります）:

{
  "errorMessage": "'requestContext'",
  "errorType": "KeyError",
  "requestId": "11111111-2222-3333-4444-555555555555",
  "stackTrace": [
    "  File \".../src/lambda_function.py\", line 15, in lambda_handler\n    http = event[\"requestContext\"][\"http\"]\n"
  ]
}

初期化（import や handler の解決）で失敗する Runtime.ImportModuleError / Runtime.HandlerNotFound では requestId が空文字で、CloudWatch には先頭に LAMBDA_WARNING: Unhandled exception. The most likely cause is an issue in the function code. ... という長い警告行が付きます。この警告が出たら「コードか設定」をまず疑う、と覚えておくと切り分けが早くなります。

よくある質問

Q. lambda_handler という名前は固定ですか？ A. 固定ではありません。ハンドラ設定が ファイル名.関数名 と一致していれば app.main でも動きます。コンソールの既定値が lambda_function.lambda_handler なので、迷ったらこの名前に揃えるのが安全です。

Q. ローカルの Python が 3.11 でも 3.14 ランタイム向けのテストはできますか？ A. できますが、uv python install 3.14 でランタイムと同じ系列を入れて揃えるのがおすすめです。構文や標準ライブラリの差に手元で気づけます。

Q. moto で Lambda の呼び出し（invoke）そのものもテストできますか？ A. moto の Lambda invoke は内部で Docker を使うため、Docker が無い環境では動きません（この記事でも未実行）。handler を直接呼ぶ方式で同じロジックを検証し、Lambda サービスとしての挙動（タイムアウト、同時実行）は sam local invoke や実環境のテストイベントで確認する、という分担が現実的です。

Q. boto3 を自分の zip に入れる必要はありますか？ A. ランタイムに同梱されているので通常は不要です。新しい API や特定バージョンの機能が必要なときだけ pip install --target で同梱します。同梱するとパッケージが大きくなりコールドスタートがわずかに伸びる点は意識しておきます。

まとめ

• Lambda の Python 関数は lambda_handler(event, context) を書き、ファイル名.関数名 をハンドラに設定する。 2026年10月時点のランタイムは python3.14 が中心
• event はトリガーごとに形が違う。 API Gateway v2 は body が文字列、S3 の key は URL エンコード済み、SQS は batchItemFailures で部分失敗を返す
• 環境変数の読み込みと boto3 クライアント生成は handler の外に置く。 ウォームスタートで再利用され、コールドスタートの影響を小さくできる
• logging を使うと [INFO]\t時刻\trequestId\t本文 の形で CloudWatch に出る。 JSON 形式に切り替えると検索しやすい
• pytest でフェイク context を渡せば handler は普通の関数としてテストできる。 S3 など AWS 依存は moto の mock_aws で模擬（6 件 0.53 秒で確認）
• デプロイは pip install --target → zip のルートにまとめる。 C 拡張は --platform manylinux2014_x86_64 --only-binary=:all: で Lambda 用ホイールを取得
• エラーは文で見分ける。 ImportModuleError は同梱漏れ、HandlerNotFound は名前違い、Task timed out はタイムアウト設定と外部待ちを見直す

株式会社RJCでは、サーバーレスの小さな関数から設計・テスト・運用の仕組みづくりまでを、プロジェクトの中で仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。