YAZエンジニアが運営するメディア

Phase 1  Lambda 作成（モック）→ 単体テスト
↓
Phase 2  API Gateway 接続 → GET/POST 確認
↓
Phase 3  DynamoDB 接続 → 永続化
↓
Phase 4  ローカル HTML から操作

import json

# 本物のDynamoDBの代わりに、メモリ上の辞書（Dict）をデータベースとして使用
# 構造: { "ユーザー名": ["タスク1", "タスク2", ...] }
MOCK_DYNAMODB = {
    "yamada": ["AWSの勉強をする", "日報を書く"],
    "tanaka": ["Pythonのコードレビュー"],
}

def lambda_handler(event, context):
    """
    API Gatewayからのリクエストを処理するハンドラー
    """
    # 1. API Gatewayからのリクエスト情報を取得
    # ※API Gatewayの**Lambda統合プロキシ**を想定
    http_method = event.get("httpMethod")
    path_parameters = event.get("pathParameters") or {}
    query_parameters = event.get("queryStringParameters") or {}
    
    # ルーティング処理
    try:
        # 【タスク取得】GET /todo?user=ユーザー名
        if http_method == "GET":
            username = query_parameters.get("user")
            if not username:
                return create_response(400, {"message": "ユーザー名（user）を指定してください"})
            
            # DynamoDB（モック）からユーザー名(PK)で検索
            tasks = MOCK_DYNAMODB.get(username, [])
            return create_response(200, {"user": username, "tasks": tasks})

        # 【タスク追加】POST /todo
        elif http_method == "POST":
            # ボディの解析
            body = json.loads(event.get("body", "{}"))
            username = body.get("user")
            task = body.get("task")
            
            if not username or not task:
                return create_response(400, {"message": "user と task は必須です"})
            
            # DynamoDB（モック）へデータ保存（ユーザーが未登録なら新規作成）
            if username not in MOCK_DYNAMODB:
                MOCK_DYNAMODB[username] = []
            MOCK_DYNAMODB[username].append(task)
            
            return create_response(201, {
                "message": "タスクを追加しました",
                "current_tasks": MOCK_DYNAMODB[username]
            })

        # 対応していないメソッド
        else:
            return create_response(405, {"message": f"Method {http_method} Not Allowed"})

    except Exception as e:
        return create_response(500, {"message": f"Internal Server Error: {str(e)}"})


def create_response(status_code, body_content):
    """
    API Gateway（プロキシ統合）が認識できる形式のレスポンスを作成するヘルパー関数
    CORS対策のヘッダーもあらかじめ付与
    """
    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",  # フロントエンド（HTML）からのクロスドメイン通信を許可
            "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type"
        },
        "body": json.dumps(body_content, ensure_ascii=False)
    }

{
  "httpMethod": "GET",
  "queryStringParameters": {
    "user": "yamada"
  }
}

{
  "httpMethod": "POST",
  "body": "{\"user\": \"yamada\", \"task\": \"課題を終わらせる\"}"
}

curl -X POST https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/todo \
  -H "Content-Type: application/json" \
  -d '{"user": "yamada", "task": "本物のDynamoDBに保存する"}'

curl -X POST https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/todo ^
  -H "Content-Type: application/json" ^
  -d "{\"user\": \"yamada\", \"task\": \"本物のDynamoDBに保存する\"}"

import json
import boto3
from datetime import datetime, timezone, timedelta
from boto3.dynamodb.conditions import Key

# DynamoDBオブジェクトの初期化
dynamodb = boto3.resource('dynamodb')
TABLE_NAME = 'ojt-dynamo-name-01'  # ← 自身で作成したテーブル名に変更
table = dynamodb.Table(TABLE_NAME)

def lambda_handler(event, context):
    """
    PK(user) と SK(timestamp) を使用した、実際のDynamoDB接続コード
    """
    http_method = event.get("httpMethod")
    query_parameters = event.get("queryStringParameters") or {}
    
    try:
        # 【タスク取得】GET /todo?user=ユーザー名
        if http_method == "GET":
            username = query_parameters.get("user")
            if not username:
                return create_response(400, {"message": "ユーザー名（user）を指定してください"})
            
            # ソートキーがある場合、get_itemではなく**query**を使う
            # 指定したユーザー(PK)のデータを全件検索
            response = table.query(
                KeyConditionExpression=Key('user').eq(username)
            )
            
            items = response.get('Items', [])
            
            # フロントエンド（JS）が扱いやすいように配列にして返す
            tasks_list = []
            for item in items:
                tasks_list.append({
                    "timestamp": item.get("timestamp"),
                    "task": item.get("task")
                })
            
            return create_response(200, {"user": username, "tasks": tasks_list})

        # 【タスク追加】POST /todo
        elif http_method == "POST":
            body = json.loads(event.get("body", "{}"))
            username = body.get("user")
            task = body.get("task")
            
            if not username or not task:
                return create_response(400, {"message": "user と task は必須です"})
            
            # 現在時刻をソートキー(SK)にする（重複を防ぎ、並び替えを容易にするため）
            # 日本時間（JST）のタイムスタンプ文字列（例: 2026-07-07T12:00:00）
            jst = timezone(timedelta(hours=+9))
            now_str = datetime.now(jst).isoformat()
            
            # PK(user)とSK(timestamp)の組み合わせが新しいので、上書きされずに別データとして保存
            table.put_item(
                Item={
                    'user': username,
                    'timestamp': now_str,
                    'task': task
                }
            )
            
            return create_response(201, {
                "message": "タスクを追加しました",
                "added_task": {
                    "user": username,
                    "timestamp": now_str,
                    "task": task
                }
            })

        # 対応していないメソッド
        else:
            return create_response(405, {"message": f"Method {http_method} Not Allowed"})

    except Exception as e:
        return create_response(500, {"message": f"DynamoDB Error: {str(e)}"})


def create_response(status_code, body_content):
    """API Gateway（プロキシ統合）用レスポンス作成ヘルパー"""
    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type"
        },
        "body": json.dumps(body_content, ensure_ascii=False)
    }

{
  "httpMethod": "POST",
  "body": "{\"user\": \"yamada\", \"task\": \"本物のDynamoDBに保存する\"}"
}

<!DOCTYPE html>
<html lang="ja">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>サーバーレス Todoリスト</title>
</head>
<body>
<div class="container">
    <h1>📝 Todo リスト</h1>

    <!-- ユーザー名入力 + 検索ボタン -->
    <div class="form-group">
        <input type="text" id="username" placeholder="ユーザー名を入力（例: yamada）">
        <button class="search-btn" onclick="fetchTodos()">検索</button>
    </div>

    <hr>

    <!-- タスク入力 + 追加ボタン -->
    <div class="form-group">
        <input type="text" id="taskContent" placeholder="新しいタスクを入力">
        <button onclick="addTodo()">追加</button>
    </div>

    <ul id="todoList"></ul>
</div>

<script>
// TODO: あなたのAPI GatewayのURL（/todoまで）に書き換えてください
const API_URL = "【ここにあなたのAPI GatewayのURLを貼り付ける】";

// 1. 【GET】ユーザー名で検索してタスク一覧を取得する関数
async function fetchTodos() {
    const username = document.getElementById("username").value.trim();
    if (!username) {
        alert("ユーザー名を入力してください");
        return;
    }
    try {
        const response = await fetch(`${API_URL}?user=${encodeURIComponent(username)}`);
        if (!response.ok) throw new Error("データの取得に失敗しました");
        const data = await response.json();
        displayTodos(data.tasks);
    } catch (error) {
        alert("エラーが発生しました: " + error.message);
    }
}

// 2. 【POST】新しいタスクを追加する関数
async function addTodo() {
    const username = document.getElementById("username").value.trim();
    const taskContent = document.getElementById("taskContent").value.trim();
    if (!username || !taskContent) {
        alert("ユーザー名とタスクの両方を入力してください");
        return;
    }
    try {
        const response = await fetch(API_URL, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ user: username, task: taskContent })
        });
        if (!response.ok) throw new Error("タスクの追加に失敗しました");
        document.getElementById("taskContent").value = "";
        fetchTodos();
    } catch (error) {
        alert("エラーが発生しました: " + error.message);
    }
}

// 3. タスク配列を画面（HTML）に描画するヘルパー関数
function displayTodos(tasks) {
    const todoListUl = document.getElementById("todoList");
    todoListUl.innerHTML = "";
    if (tasks.length === 0) {
        todoListUl.innerHTML = "<li>タスクはありません</li>";
        return;
    }
    tasks.forEach(item => {
        const li = document.createElement("li");
        const displayTime = item.timestamp
            ? item.timestamp.substring(0, 16).replace("T", " ")
            : "";
        li.innerHTML = `
            <span>${escapeHtml(item.task)}</span>
            <span class="timestamp">${displayTime}</span>
        `;
        todoListUl.appendChild(li);
    });
}

// 4. セキュリティ対策（XSS対策）：文字エスケープ関数
function escapeHtml(str) {
    return str.replace(/[&<>"']/g, match => {
        const escape = { '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": ''' };
        return escape[match];
    });
}
</script>
</body>
</html>