CDKで単純なCRUD APIを作ってみた。DynamoDB操作用のLambda5種類(レコード作成・読込・更新・削除・一覧)を作ってAPIゲートウェイ経由で操作する。

# 任意の作業ディレクトリで
mkdir mini-cdk-crud && cd mini-cdk-crud
# CDKプロジェクト生成
cdk init app --language typescript
npm i aws-cdk-lib constructs

# lambdaファイル作成
mkdir -p src/app
touch src/app/common.py
touch src/app/handler_create.py
touch src/app/handler_delete.py
touch src/app/handler_list.py
touch src/app/handler_read.py
touch src/app/handler_update.py

# 任意の作業ディレクトリで

mkdir mini-cdk-crud && cd mini-cdk-crud

# CDKプロジェクト生成

cdk init app --language typescript

npm i aws-cdk-lib constructs

# lambdaファイル作成

mkdir -p src/app

touch src/app/common.py

touch src/app/handler_create.py

touch src/app/handler_delete.py

touch src/app/handler_list.py

touch src/app/handler_read.py

touch src/app/handler_update.py

2-1) 共通ユーティリティ（任意・楽になる）src/app/common.py

import json
import os
import boto3
from botocore.exceptions import ClientError
from typing import Any, Dict

# DynamoDB テーブル（環境変数 TABLE_NAME を使用）
_dynamodb = boto3.resource("dynamodb")
_table = _dynamodb.Table(os.environ["TABLE_NAME"])

def ok(body: Dict[str, Any], status: int = 200):
    """JSONレスポンス（成功）"""
    return {
        "statusCode": status,
        "headers": {"content-type": "application/json"},
        "body": json.dumps(body, ensure_ascii=False),
    }

def ng(message: str, status: int):
    """JSONレスポンス（エラー）"""
    return ok({"error": message}, status=status)

def get_table():
    """テーブル取得（テストしやすく分離）"""
    return _table

def get_path_id(event) -> str | None:
    """API Gateway のパスパラメータから id を取得"""
    path_params = event.get("pathParameters") or {}
    return path_params.get("id")

import json

import os

import boto3

from botocore.exceptions import ClientError

from typing import Any, Dict

# DynamoDB テーブル（環境変数 TABLE_NAME を使用）

_dynamodb = boto3.resource("dynamodb")

_table = _dynamodb.Table(os.environ["TABLE_NAME"])

def ok(body: Dict[str, Any], status: int = 200):

"""JSONレスポンス（成功）"""

return {

"statusCode": status,

"headers": {"content-type": "application/json"},

"body": json.dumps(body, ensure_ascii=False),

}

def ng(message: str, status: int):

"""JSONレスポンス（エラー）"""

return ok({"error": message}, status=status)

def get_table():

"""テーブル取得（テストしやすく分離）"""

return _table

def get_path_id(event) -> str | None:

"""API Gateway のパスパラメータから id を取得"""

path_params = event.get("pathParameters") or {}

return path_params.get("id")

2-2) 作成（POST /items）src/app/handler_create.py

import json
import uuid
from datetime import datetime, timezone
from common import ok, ng, get_table

def handler(event, context):
    # リクエストボディ(JSON)を辞書に
    try:
        body = json.loads(event.get("body") or "{}")
    except json.JSONDecodeError:
        return ng("不正なJSONです", 400)

    name = body.get("name")
    if not name:
        return ng("name は必須です", 400)

    item = {
        "id": str(uuid.uuid4()),
        "name": name,
        "createdAt": datetime.now(timezone.utc).isoformat(),
    }

    table = get_table()
    table.put_item(Item=item)

    return ok(item, 201)  # 201 Created

import json

import uuid

from datetime import datetime, timezone

from common import ok, ng, get_table

def handler(event, context):

# リクエストボディ(JSON)を辞書に

try:

body = json.loads(event.get("body") or "{}")

except json.JSONDecodeError:

return ng("不正なJSONです", 400)

name = body.get("name")

if not name:

return ng("name は必須です", 400)

item = {

"id": str(uuid.uuid4()),

"name": name,

"createdAt": datetime.now(timezone.utc).isoformat(),

}

table = get_table()

table.put_item(Item=item)

return ok(item, 201) # 201 Created

2-3) 取得（GET /items/{id}）src/app/handler_read.py

from common import ok, ng, get_table, get_path_id

def handler(event, context):
    item_id = get_path_id(event)
    if not item_id:
        return ng("id が必要です", 400)

    table = get_table()
    resp = table.get_item(Key={"id": item_id})
    item = resp.get("Item")
    if not item:
        return ng("対象が見つかりません", 404)

    return ok(item)  # 200

from common import ok, ng, get_table, get_path_id

def handler(event, context):

item_id = get_path_id(event)

if not item_id:

return ng("id が必要です", 400)

table = get_table()

resp = table.get_item(Key={"id": item_id})

item = resp.get("Item")

if not item:

return ng("対象が見つかりません", 404)

return ok(item) # 200

2-4) 更新（PUT /items/{id}）src/app/handler_update.py

import json
from datetime import datetime, timezone
from common import ok, ng, get_table, get_path_id

def handler(event, context):
    item_id = get_path_id(event)
    if not item_id:
        return ng("id が必要です", 400)

    try:
        body = json.loads(event.get("body") or "{}")
    except json.JSONDecodeError:
        return ng("不正なJSONです", 400)

    # 今回は name のみ更新（必要に応じて項目を増やす）
    if "name" not in body:
        return ng("更新項目がありません（name を指定してください）", 400)

    table = get_table()
    resp = table.update_item(
        Key={"id": item_id},
        UpdateExpression="SET #n = :name, updatedAt = :u",
        ExpressionAttributeNames={"#n": "name"},
        ExpressionAttributeValues={
            ":name": body["name"],
            ":u": datetime.now(timezone.utc).isoformat(),
        },
        ReturnValues="ALL_NEW",
    )

    return ok(resp.get("Attributes", {}))  # 更新後の全項目を返却

import json

from datetime import datetime, timezone

from common import ok, ng, get_table, get_path_id

def handler(event, context):

item_id = get_path_id(event)

if not item_id:

return ng("id が必要です", 400)

try:

body = json.loads(event.get("body") or "{}")

except json.JSONDecodeError:

return ng("不正なJSONです", 400)

# 今回は name のみ更新（必要に応じて項目を増やす）

if "name" not in body:

return ng("更新項目がありません（name を指定してください）", 400)

table = get_table()

resp = table.update_item(

Key={"id": item_id},

UpdateExpression="SET #n = :name, updatedAt = :u",

ExpressionAttributeNames={"#n": "name"},

ExpressionAttributeValues={

":name": body["name"],

":u": datetime.now(timezone.utc).isoformat(),

ReturnValues="ALL_NEW",

)

return ok(resp.get("Attributes", {})) # 更新後の全項目を返却

2-5) 削除（DELETE /items/{id}）src/app/handler_delete.py

from common import ok, ng, get_table, get_path_id

def handler(event, context):
    item_id = get_path_id(event)
    if not item_id:
        return ng("id が必要です", 400)

    table = get_table()
    # 先に存在チェック（なくても delete は200だが、学習用に404も返せるように）
    exist = table.get_item(Key={"id": item_id}).get("Item")
    if not exist:
        return ng("対象が見つかりません", 404)

    table.delete_item(Key={"id": item_id})
    return ok({"deleted": item_id})

from common import ok, ng, get_table, get_path_id

def handler(event, context):

item_id = get_path_id(event)

if not item_id:

return ng("id が必要です", 400)

table = get_table()

# 先に存在チェック（なくても delete は200だが、学習用に404も返せるように）

exist = table.get_item(Key={"id": item_id}).get("Item")

if not exist:

return ng("対象が見つかりません", 404)

table.delete_item(Key={"id": item_id})

return ok({"deleted": item_id})

2-6) 一覧（GET /）src/app/handler_list.py

import os
import json
from decimal import Decimal
from typing import Any, Dict, List, Tuple, Optional
from common import ok, ng, get_table

def _parse_limit(event, default=20, max_limit=100) -> int:
    """limit クエリをパースして上限をかける"""
    qs = event.get("queryStringParameters") or {}
    try:
        v = int(qs.get("limit", default))
    except (TypeError, ValueError):
        v = default
    return max(1, min(v, max_limit))

def _parse_token(event) -> Optional[Dict[str, Any]]:
    """token（JSONエンコードされた LastEvaluatedKey）をデコード"""
    qs = event.get("queryStringParameters") or {}
    token = qs.get("token")
    if not token:
        return None
    try:
        return json.loads(token)
    except json.JSONDecodeError:
        return None

def _parse_name_filter(event) -> Optional[str]:
    """name クエリ（部分一致用）。例: ?name=abc"""
    qs = event.get("queryStringParameters") or {}
    name = qs.get("name")
    if name:
        name = name.strip()
        if name:
            return name
    return None

def _dynamodb_to_json(obj):
    """DynamoDBが返すDecimalなどをJSONにできる形に変換"""
    if isinstance(obj, list):
        return [_dynamodb_to_json(x) for x in obj]
    if isinstance(obj, dict):
        return {k: _dynamodb_to_json(v) for k, v in obj.items()}
    if isinstance(obj, Decimal):
        # 数値は float に変換（必要なら str にする運用もある）
        return float(obj)
    return obj

def handler(event, context):
    table = get_table()

    limit = _parse_limit(event)
    eks   = _parse_token(event)       # ExclusiveStartKey
    name  = _parse_name_filter(event) # 簡易フィルタ（部分一致）

    scan_kwargs: Dict[str, Any] = {
        "Limit": limit,
    }
    if eks:
        scan_kwargs["ExclusiveStartKey"] = eks

    # name フィルタ（FilterExpression / contains）。※フルスキャン後にフィルタのため費用/性能は用途に応じて要注意
    if name:
        from boto3.dynamodb.conditions import Attr
        scan_kwargs["FilterExpression"] = Attr("name").contains(name)

    resp = table.scan(**scan_kwargs)

    items = _dynamodb_to_json(resp.get("Items", []))
    next_token = resp.get("LastEvaluatedKey")
    body = {
        "items": items,
        "count": len(items),
        "nextToken": json.dumps(next_token) if next_token else None,
    }
    return ok(body)

import os

import json

from decimal import Decimal

from typing import Any, Dict, List, Tuple, Optional

from common import ok, ng, get_table

def _parse_limit(event, default=20, max_limit=100) -> int:

"""limit クエリをパースして上限をかける"""

qs = event.get("queryStringParameters") or {}

try:

v = int(qs.get("limit", default))

except (TypeError, ValueError):

v = default

return max(1, min(v, max_limit))

def _parse_token(event) -> Optional[Dict[str, Any]]:

"""token（JSONエンコードされた LastEvaluatedKey）をデコード"""

qs = event.get("queryStringParameters") or {}

token = qs.get("token")

if not token:

return None

try:

return json.loads(token)

except json.JSONDecodeError:

return None

def _parse_name_filter(event) -> Optional[str]:

"""name クエリ（部分一致用）。例: ?name=abc"""

qs = event.get("queryStringParameters") or {}

name = qs.get("name")

if name:

name = name.strip()

if name:

return name

return None

def _dynamodb_to_json(obj):

"""DynamoDBが返すDecimalなどをJSONにできる形に変換"""

if isinstance(obj, list):

return [_dynamodb_to_json(x) for x in obj]

if isinstance(obj, dict):

return {k: _dynamodb_to_json(v) for k, v in obj.items()}

if isinstance(obj, Decimal):

# 数値は float に変換（必要なら str にする運用もある）

return float(obj)

return obj

def handler(event, context):

table = get_table()

limit = _parse_limit(event)

eks = _parse_token(event) # ExclusiveStartKey

name = _parse_name_filter(event) # 簡易フィルタ（部分一致）

scan_kwargs: Dict[str, Any] = {

"Limit": limit,

}

if eks:

scan_kwargs["ExclusiveStartKey"] = eks

# name フィルタ（FilterExpression / contains）。※フルスキャン後にフィルタのため費用/性能は用途に応じて要注意

if name:

from boto3.dynamodb.conditions import Attr

scan_kwargs["FilterExpression"] = Attr("name").contains(name)

resp = table.scan(**scan_kwargs)

items = _dynamodb_to_json(resp.get("Items", []))

next_token = resp.get("LastEvaluatedKey")

body = {

"items": items,

"count": len(items),

"nextToken": json.dumps(next_token) if next_token else None,

}

return ok(body)

次ににインフラ部分を定義
DynamoDBテーブル作成
5つのLambda（Create / Read / Update / Delete / List）
API Gateway 作成(API URL出力)

lib/mini-cdk-crud-stack.ts

// コメントはすべて日本語
import {
  Stack,
  StackProps,
  Duration,
  CfnOutput,
  RemovalPolicy,
} from "aws-cdk-lib";
import { Construct } from "constructs";
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as apigw from "aws-cdk-lib/aws-apigateway";
import * as dynamodb from "aws-cdk-lib/aws-dynamodb";
import * as path from "path";

export class MiniCdkCrudStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    // --- DynamoDBテーブル作成（学習用なので削除許可） ---
    const table = new dynamodb.Table(this, "ItemsTable", {
      tableName: "items",
      partitionKey: { name: "id", type: dynamodb.AttributeType.STRING },
      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST, // 従量課金で楽
      removalPolicy: RemovalPolicy.DESTROY, // 学習用。実運用は RETAIN を検討
    });

    // --- Lambda 共通設定 ---
    const commonProps: Omit<lambda.FunctionProps, "handler" | "code"> = {
      runtime: lambda.Runtime.PYTHON_3_12,
      timeout: Duration.seconds(10),
      environment: {
        TABLE_NAME: table.tableName,
      },
    };

    // --- 5つのLambda（Create / Read / Update / Delete / List） ---
    const codeFromAsset = lambda.Code.fromAsset(
      path.join(__dirname, "../src/app")
    );

    const fnCreate = new lambda.Function(this, "CreateFn", {
      ...commonProps,
      handler: "handler_create.handler", // ファイル名.関数名
      code: codeFromAsset,
      description: "データ作成（POST /items）",
    });

    const fnRead = new lambda.Function(this, "ReadFn", {
      ...commonProps,
      handler: "handler_read.handler",
      code: codeFromAsset,
      description: "データ取得（GET /items/{id}）",
    });

    const fnUpdate = new lambda.Function(this, "UpdateFn", {
      ...commonProps,
      handler: "handler_update.handler",
      code: codeFromAsset,
      description: "データ更新（PUT /items/{id}）",
    });

    const fnDelete = new lambda.Function(this, "DeleteFn", {
      ...commonProps,
      handler: "handler_delete.handler",
      code: codeFromAsset,
      description: "データ削除（DELETE /items/{id}）",
    });
    // 一覧取得（GET /items）
    const fnList = new lambda.Function(this, "ListFn", {
      ...commonProps,
      handler: "handler_list.handler",
      code: codeFromAsset,
      description: "一覧取得（GET /items）",
    });

    // --- DDBへの権限付与（読み書き） ---
    table.grantReadWriteData(fnCreate);
    table.grantReadWriteData(fnRead);
    table.grantReadWriteData(fnUpdate);
    table.grantReadWriteData(fnDelete);
    table.grantReadData(fnList);

    // --- API Gateway 作成 ---
    const api = new apigw.RestApi(this, "CrudApi", {
      restApiName: "mini-cdk-crud-api",
      deployOptions: { stageName: "prod" },
      // 簡易CORS（必要な場合）
      defaultCorsPreflightOptions: {
        allowOrigins: apigw.Cors.ALL_ORIGINS,
        allowMethods: ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
      },
    });

    // /items リソース（POST）
    const items = api.root.addResource("items");
    items.addMethod("POST", new apigw.LambdaIntegration(fnCreate));
    // GET /items → 一覧取得
    items.addMethod("GET", new apigw.LambdaIntegration(fnList));

    // /items/{id} リソース（GET/PUT/DELETE）
    const item = items.addResource("{id}");
    item.addMethod("GET", new apigw.LambdaIntegration(fnRead));
    item.addMethod("PUT", new apigw.LambdaIntegration(fnUpdate));
    item.addMethod("DELETE", new apigw.LambdaIntegration(fnDelete));

    // --- 出力（URLとテーブル名） ---
    new CfnOutput(this, "ApiBaseUrl", {
      value: api.url ?? "undefined",
      description: "APIのベースURL",
    });
    new CfnOutput(this, "TableName", {
      value: table.tableName,
      description: "DynamoDBテーブル名",
    });
  }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

// コメントはすべて日本語

import {

Stack,

StackProps,

Duration,

CfnOutput,

RemovalPolicy,

} from "aws-cdk-lib";

import { Construct } from "constructs";

import * as lambda from "aws-cdk-lib/aws-lambda";

import * as apigw from "aws-cdk-lib/aws-apigateway";

import * as dynamodb from "aws-cdk-lib/aws-dynamodb";

import * as path from "path";

export class MiniCdkCrudStack extends Stack {

constructor(scope: Construct, id: string, props?: StackProps) {

super(scope, id, props);

// --- DynamoDBテーブル作成（学習用なので削除許可） ---

const table = new dynamodb.Table(this, "ItemsTable", {

tableName: "items",

partitionKey: { name: "id", type: dynamodb.AttributeType.STRING },

billingMode: dynamodb.BillingMode.PAY_PER_REQUEST, // 従量課金で楽

removalPolicy: RemovalPolicy.DESTROY, // 学習用。実運用は RETAIN を検討

});

// --- Lambda 共通設定 ---

const commonProps: Omit<lambda.FunctionProps, "handler" | "code"> = {

runtime: lambda.Runtime.PYTHON_3_12,

timeout: Duration.seconds(10),

environment: {

TABLE_NAME: table.tableName,

};

// --- 5つのLambda（Create / Read / Update / Delete / List） ---

const codeFromAsset = lambda.Code.fromAsset(

path.join(__dirname, "../src/app")

);

const fnCreate = new lambda.Function(this, "CreateFn", {

...commonProps,

handler: "handler_create.handler", // ファイル名.関数名

code: codeFromAsset,

description: "データ作成（POST /items）",

});

const fnRead = new lambda.Function(this, "ReadFn", {

...commonProps,

handler: "handler_read.handler",

code: codeFromAsset,

description: "データ取得（GET /items/{id}）",

});

const fnUpdate = new lambda.Function(this, "UpdateFn", {

...commonProps,

handler: "handler_update.handler",

code: codeFromAsset,

description: "データ更新（PUT /items/{id}）",

});

const fnDelete = new lambda.Function(this, "DeleteFn", {

...commonProps,

handler: "handler_delete.handler",

code: codeFromAsset,

description: "データ削除（DELETE /items/{id}）",

});

// 一覧取得（GET /items）

const fnList = new lambda.Function(this, "ListFn", {

...commonProps,

handler: "handler_list.handler",

code: codeFromAsset,

description: "一覧取得（GET /items）",

});

// --- DDBへの権限付与（読み書き） ---

table.grantReadWriteData(fnCreate);

table.grantReadWriteData(fnRead);

table.grantReadWriteData(fnUpdate);

table.grantReadWriteData(fnDelete);

table.grantReadData(fnList);

// --- API Gateway 作成 ---

const api = new apigw.RestApi(this, "CrudApi", {

restApiName: "mini-cdk-crud-api",

deployOptions: { stageName: "prod" },

// 簡易CORS（必要な場合）

defaultCorsPreflightOptions: {

allowOrigins: apigw.Cors.ALL_ORIGINS,

allowMethods: ["GET", "POST", "PUT", "DELETE", "OPTIONS"],

});

// /items リソース（POST）

const items = api.root.addResource("items");

items.addMethod("POST", new apigw.LambdaIntegration(fnCreate));

// GET /items → 一覧取得

items.addMethod("GET", new apigw.LambdaIntegration(fnList));

// /items/{id} リソース（GET/PUT/DELETE）

const item = items.addResource("{id}");

item.addMethod("GET", new apigw.LambdaIntegration(fnRead));

item.addMethod("PUT", new apigw.LambdaIntegration(fnUpdate));

item.addMethod("DELETE", new apigw.LambdaIntegration(fnDelete));

// --- 出力（URLとテーブル名） ---

new CfnOutput(this, "ApiBaseUrl", {

value: api.url ?? "undefined",

description: "APIのベースURL",

});

new CfnOutput(this, "TableName", {

value: table.tableName,

description: "DynamoDBテーブル名",

});

}

# デプロイ 
cdk bootstrap 
cdk deploy

# デプロイ

cdk bootstrap

cdk deploy

コマンドラインから、API操作してみる。

# ベースURL（あなたの出力に合わせて設定）
BASE="https://7vf5hfu5b0.execute-api.ap-northeast-1.amazonaws.com/prod/"

# 1) 作成（POST /items）
CREATE_RES=$(curl -s -X POST "${BASE}items" \
  -H "content-type: application/json" \
  -d '{"name":"first item"}')
echo "${CREATE_RES}"

# 作成レスポンスから id を取り出して使い回し
ID=$(python -c "import sys, json; print(json.load(sys.stdin)['id'])" <<< "${CREATE_RES}")
echo "ID=${ID}"

# 2) 取得（GET /items/{id}）
curl -s "${BASE}items/${ID}"

# 3) 更新（PUT /items/{id}）
curl -s -X PUT "${BASE}items/${ID}" \
  -H "content-type: application/json" \
  -d '{"name":"renamed"}'

# 4) 削除（DELETE /items/{id}）
curl -s -X DELETE "${BASE}items/${ID}"

# ベースURL（あなたの出力に合わせて設定）

BASE="https://7vf5hfu5b0.execute-api.ap-northeast-1.amazonaws.com/prod/"

# 1) 作成（POST /items）

CREATE_RES=$(curl -s -X POST "${BASE}items" \

-H "content-type: application/json" \

-d '{"name":"first item"}')

echo "${CREATE_RES}"

# 作成レスポンスから id を取り出して使い回し

ID=$(python -c "import sys, json; print(json.load(sys.stdin)['id'])" <<< "${CREATE_RES}")

echo "ID=${ID}"

# 2) 取得（GET /items/{id}）

curl -s "${BASE}items/${ID}"

# 3) 更新（PUT /items/{id}）

curl -s -X PUT "${BASE}items/${ID}" \

-H "content-type: application/json" \

-d '{"name":"renamed"}'

# 4) 削除（DELETE /items/{id}）

curl -s -X DELETE "${BASE}items/${ID}"

一覧表示とページング

# まず2件ほど作成
curl -s -X POST "${BASE}items" -H "content-type: application/json" -d '{"name":"apple"}'  | jq .
curl -s -X POST "${BASE}items" -H "content-type: application/json" -d '{"name":"banana"}' | jq .

# 一覧（最初のページ）
curl -s "${BASE}items" | jq .

# 10件ずつ（limit）
curl -s "${BASE}items?limit=10" | jq .

# 名前に "a" を含む（簡易フィルタ）
curl -s "${BASE}items?name=a" | jq .

# 続きのページ（nextToken を使う）
RES=$(curl -s "${BASE}items?limit=1")
TOKEN=$(echo "$RES" | jq -r '.nextToken' | sed 's/"//g')
curl -s "${BASE}items?limit=1&token=${TOKEN}" | jq .

# まず2件ほど作成

curl -s -X POST "${BASE}items" -H "content-type: application/json" -d '{"name":"apple"}' | jq .

curl -s -X POST "${BASE}items" -H "content-type: application/json" -d '{"name":"banana"}' | jq .

# 一覧（最初のページ）

curl -s "${BASE}items" | jq .

# 10件ずつ（limit）

curl -s "${BASE}items?limit=10" | jq .

# 名前に "a" を含む（簡易フィルタ）

curl -s "${BASE}items?name=a" | jq .

# 続きのページ（nextToken を使う）

RES=$(curl -s "${BASE}items?limit=1")

TOKEN=$(echo "$RES" | jq -r '.nextToken' | sed 's/"//g')

curl -s "${BASE}items?limit=1&token=${TOKEN}" | jq .

関連記事