Python × AWS Lambda 実践入門｜ユースケース別スニペット集（boto3の第一歩）

「サーバーを持たずにコードを動かす」AWS Lambda を、Python と boto3 で触ってみるための実践ガイドです。未経験〜駆け出しの方向けに、最小のハンドラから、S3トリガー・定期実行・通知・簡易APIまで、コピペできる断片と一緒にユースケース別で整理しました。使う・使わないの判断と、事故らないコツも添えます。

01Lambda の考え方 — イベント駆動と「使う／使わない」

AWS Lambda は「サーバーを用意せずに、イベントが起きたときだけコードを動かす」しくみです。誰かが S3 にファイルを置いた、決まった時刻になった、API にリクエストが来た――そうしたイベントをきっかけに関数が起動し、処理が終われば止まります。常駐サーバの起動・パッチ当て・スケーリングを自分で抱えなくてよいのが最大の利点です。

向いているのは、次のような「イベント駆動で、上限時間内に終わる」処理です。逆に、常時稼働が要る・1回が長時間・低レイテンシが命、という処理は別の選択肢が向きます。

• 向く：ファイルアップロードの後処理、日次の軽いバッチ、通知の送信、軽量な API、他サービスをつなぐ「のりしろ」処理
• 向きにくい：常時起動のサーバ、15分を超える長時間処理、大きな常駐メモリが要る処理、数ミリ秒を争う超低レイテンシ

料金は「呼ばれた回数」と「実行時間×メモリ」に応じた従量課金です。使わなければ基本的に費用は発生しません。裏を返すと、無限ループや過剰なリトライは「動いた分だけ」コストと時間を食うので、暴走しない設計が大切です（具体的な単価は変わるため、最新は公式の料金ページで確認してください）。

OS やネットワークの基礎に不安があれば クラウドエンジニアの教科書（TCP/IP） も合わせてどうぞ。なお、常時起動が必要なプロセスは Lambda ではなく systemd で常駐サービス化するほうが向く場面もあります（サーバーレスとの対比として押さえておくと判断が速くなります）。

02最小の Lambda — handler・ログ・環境変数・タイムアウト

Lambda 関数の入り口はハンドラ関数です。Python では def lambda_handler(event, context): の形が基本で、Lambda が呼び出すたびにこの関数が実行されます。event は呼び出し元から渡るデータ（dict）、context は実行に関する情報（残り時間など）です。

import os
import json
import logging

# ハンドラの「外」で初期化しておくと、実行環境が再利用された2回目以降が速い
logger = logging.getLogger()
logger.setLevel("INFO")             # print でもログは出るが、レベル管理できる logger を推奨
TABLE_NAME = os.environ.get("TABLE_NAME")   # 環境変数は os.environ で読む

def lambda_handler(event, context):
    # context から実行の残り時間(ミリ秒)などが取れる
    logger.info(f"start: remaining_ms={context.get_remaining_time_in_millis()}")
    name = event.get("name", "world")
    return {"message": f"hello, {name}"}

ハンドラ名は「ファイル名.関数名」で指定します。上のコードを lambda_function.py に置いたなら、ハンドラは lambda_function.lambda_handler です（コンソール既定値と同じ）。

• ログ：print() でも CloudWatch Logs に出ますが、logging を使うとレベル（INFO/ERROR）で絞れて運用が楽になります。
• 環境変数：接続先やバケット名などは os.environ で受け取り、コードに直書きしません。
• タイムアウト／メモリ：タイムアウトは最大15分。メモリは最小128MBから設定でき、メモリを増やすと CPU も比例して増えるため、遅い処理はメモリ増で速くなることがあります。

03boto3 の基本 — client と resource・認証・最小権限

boto3 は AWS を Python から操作する SDK です。呼び出し方には client（低レベル）と resource（高レベル）の2種類があります。

import boto3

# client: 低レベル。ほぼ全 API をカバーし、新機能もここに入る（基本はこちら推奨）
s3 = boto3.client("s3")
resp = s3.list_buckets()

# resource: 高レベルで書きやすいが、機能追加が凍結されている
#   → 既存コードは動くが、新しい機能は client 側にしか来ないことがある
ddb = boto3.resource("dynamodb")
table = ddb.Table("my-table")

# 認証情報はコードに書かない。Lambda は「実行ロール」の一時認証情報を
#   boto3 が自動的に使う。リージョンも環境変数 AWS_REGION から自動で入る

迷ったら client を選んでおくと無難です。resource インターフェースは機能フリーズ（新機能の追加なし。廃止ではない）の状態で、AWS 側も新規コードは client を推奨しています。

認証は実行ロール（IAM ロール）に任せます。Lambda に割り当てたロールの権限だけで boto3 が動くので、そのロールに「本当に必要な操作だけ」を与えるのが最小権限の考え方です。同じ操作を手元から試したいときは AWS CLI 実践リファレンス が対応します。

04UC：S3 トリガー — アップロードで発火して処理する

S3 にファイルが置かれたら Lambda を起動する、という定番の連携です。イベントは event["Records"] の配列で届き、そこからバケット名とオブジェクトキーを取り出します。

import urllib.parse
import boto3

s3 = boto3.client("s3")

def lambda_handler(event, context):
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        # キーは URL エンコードされている(スペースや日本語) → 必ずデコードする
        key = urllib.parse.unquote_plus(record["s3"]["object"]["key"])

        head = s3.head_object(Bucket=bucket, Key=key)
        print(f"uploaded: {bucket}/{key} size={head['ContentLength']}")
        # ここで変換・検証・別バケットへコピーなどの処理を行う

    return {"statusCode": 200}

• キーのURL デコードを忘れると、スペースや日本語を含むファイルで「見つからない」エラーになりがちです。unquote_plus を通すのが定番です。
• 1イベントに複数レコードが入ることがあるため、Records はループで処理します。
• 無限ループ注意：同じバケットに書き戻すトリガーだと、書き込みがまた発火して無限ループになります。出力は別バケット／別プレフィックスへ。

05UC：定期実行 — EventBridge Scheduler で日次バッチ

「毎日 0 時に集計する」のような定期実行は、EventBridge Scheduler から Lambda を呼ぶのが基本形です。Scheduler は rate()（間隔）、cron()（時刻指定）、at()（1回だけ）の3種類のスケジュールに対応します。

Lambda 側は、いつも通りのハンドラを書くだけです。起動タイミングは Scheduler が担当します。

# 日次バッチ用のハンドラ。中身は普通の関数。起動は Scheduler 側が受け持つ
def lambda_handler(event, context):
    # event には Scheduler で設定した任意のペイロードが入る(自分で定義できる)
    target_date = event.get("date", "today")
    # ここに集計・レポート生成・掃除などの処理を書く
    return {"status": "done", "target": target_date}

スケジュールの指定はコードではなく式で書きます。代表例は次のとおりです。

rate(1 day)            # 1日ごと
rate(5 minutes)        # 5分ごと
cron(0 15 * * ? *)     # 毎日 15:00 に実行（フィールドは 分 時 日 月 曜日 年）

• EventBridge Scheduler はタイムゾーン指定に対応するため、Asia/Tokyo を指定すれば JST 基準で書けます（旧来の EventBridge ルールは UTC 基準）。
• 起動精度は基本 60 秒単位です。ミリ秒精度の厳密なタイミングを前提にしない設計にします。
• スケジュール・ロール・関数は手作業で作らず、後述のとおり IaC で管理すると再現性が上がります。

06UC：通知 — SNS / Slack Webhook / CloudWatch アラーム

処理の完了や失敗を知らせる通知も Lambda の定番です。AWS 内へファンアウトするなら SNS publish、Slack へ直接送るなら Incoming Webhook に標準ライブラリの urllib で POST します（追加ライブラリの同梱が不要なのが利点）。

import os
import json
import urllib.request
import boto3

sns = boto3.client("sns")

def lambda_handler(event, context):
    msg = "日次バッチが完了しました"

    # (1) SNS へ publish（メール/SMS/他サービスへファンアウトできる）
    sns.publish(TopicArn=os.environ["TOPIC_ARN"], Subject="通知", Message=msg)

    # (2) Slack Incoming Webhook へ POST（Webhook URL は環境変数や Secrets で）
    body = json.dumps({"text": msg}).encode("utf-8")
    req = urllib.request.Request(
        os.environ["SLACK_WEBHOOK_URL"],
        data=body,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=5) as res:
        print("slack status:", res.status)

    return {"statusCode": 200}

監視と組み合わせるなら、CloudWatch アラーム → SNS トピック → （メール or Lambda）という連携がよく使われます。メトリクスがしきい値を超えたらアラームが SNS に通知し、SNS が担当者へメールしたり、整形用の Lambda を起動して Slack に流したりできます。

• Slack の Webhook URL は秘密情報です。コードに直書きせず、環境変数か Secrets Manager 経由にします。
• 外部への POST はタイムアウトを必ず指定します（相手が遅いと Lambda ごと固まるため）。

07UC：簡易 ETL／API — DynamoDB と API Gateway

DynamoDB への読み書きと、API Gateway 経由の軽い API を組み合わせると、小さな Web API がすぐ作れます。client の DynamoDB は型付きの値（S=文字列、N=数値）で受け渡すのが特徴です。

import json
import boto3

ddb = boto3.client("dynamodb")
TABLE = "visits"

def lambda_handler(event, context):
    # API Gateway(プロキシ統合)からの呼び出しを想定
    path_id = (event.get("pathParameters") or {}).get("id", "unknown")

    # 書き込み: 値は型付きで渡す（S=文字列, N=数値 ...）
    ddb.put_item(TableName=TABLE, Item={"id": {"S": path_id}})

    # 読み出し: 見つからなければ "Item" キー自体が無い
    got = ddb.get_item(TableName=TABLE, Key={"id": {"S": path_id}})
    item = got.get("Item")

    # プロキシ統合のレスポンスは statusCode / headers / body(文字列) の形
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"id": path_id, "found": item is not None}),
    }

• API Gateway のLambda プロキシ統合では、レスポンスは必ず statusCode と、文字列化した body を返します（dict をそのまま返すと 502 になりがち）。
• 入力の pathParameters は None のことがあるため、(... or {}) で守ってから読むと安全です。
• 型付きが面倒なら boto3.resource("dynamodb").Table(...) だと Python の型のまま書けます。ただし resource は機能フリーズ中な点を踏まえて選びます。

08現場のコツ — べき等性・リトライ・秘密・監視・IaC

小さく作った Lambda を「本番で安心して回す」ために、最低限おさえたい勘所です。

• べき等性：同じイベントが2回届いても結果が壊れないように作ります（同じ ID の処理はスキップ、上書きで統一する等）。Lambda のイベントは「少なくとも1回」届く前提で考えます。
• リトライ／DLQ：非同期呼び出しは自動でリトライされます。それでも失敗したイベントは DLQ（デッドレターキュー：SQS/SNS） に退避させ、後から調査・再処理できるようにします。
• 秘密は Secrets Manager：DB パスワードや外部トークンはコードや環境変数に直書きせず、Secrets Manager / SSM パラメータストアから実行時に取得します。
• ログ・監視：CloudWatch Logs に構造化してログを出し、エラー率やスロットリングにアラームを張ります。
• IaC で管理：関数・実行ロール・スケジュールは手作業で作らずコード管理に。Organizations／IaC の設計の考え方が土台になります。

Secrets Manager から秘密を取り出す小さな例です（取得結果は JSON 文字列のことが多い）。

import json
import boto3

sm = boto3.client("secretsmanager")

def get_secret(name):
    resp = sm.get_secret_value(SecretId=name)
    return json.loads(resp["SecretString"])   # {"user": "...", "password": "..."} など

09早見表 — よく使う boto3 スニペット逆引き＆トリガー一覧

「やりたいこと」から boto3（client）のスニペットを引ける早見表です。まずはコピペして動かし、細部は公式リファレンスで確認してください。

主なトリガーと、event から取り出すデータの入り口です。

10よくある質問（FAQ）