はじめに

AWS CDK（Python）を活用して、DynamoDB、Lambda、およびAPI Gatewayを組み合わせた柔軟性の高いCRUD（Create, Read, Update, Delete）APIを構築し、IPアドレス制限を設定する方法について詳しく解説します。
特に、DynamoDBのテーブル構成をカスタマイズしやすくするポイントに焦点を当てています。
初心者から中級者の方まで、幅広いレベルの方に役立つ内容となっています。

前提条件

以下のツールとアカウントが準備されていることを確認してください。

AWSアカウント: AWSリソースを作成するために必要です。
AWS CLI: インストールと設定が完了していること。
Python: Python 3.7以上がインストールされていること。
AWS CDK: グローバルにインストールされていること。インストールされていない場合は、以下のコマンドでインストールできます。
```
npm install -g aws-cdk
```
Node.jsとnpm: CDKの動作に必要です。インストールされていない場合は公式サイトからインストールしてください。

プロジェクトのセットアップ

1. CDKプロジェクトの初期化

まず、新しいディレクトリを作成し、CDKプロジェクトをPythonで初期化します。

mkdir cdk-crud-api-python
cd cdk-crud-api-python
cdk init app --language python

このコマンドにより、以下のようなディレクトリ構成が生成されます。

cdk-crud-api-python/
├── README.md
│   ├── # プロジェクトの概要、セットアップ方法、使用方法などのドキュメントを記載
├── app.py
│   ├── # AWS CDKアプリケーションのエントリーポイント。スタックを定義し、デプロイの設定を行う
├── cdk.json
│   ├── # CDKの設定ファイル。アプリケーションの起動方法やコンテキスト情報を定義
├── cdk_crud_api_python
│   ├── __init__.py
│   │   ├── # Pythonパッケージとして認識させるための初期化ファイル
│   └── cdk_crud_api_python_stack.py
│       ├── # AWSリソース（例：API Gateway、Lambda、DynamoDBなど）を定義するCDKスタックの実装
├── requirements-dev.txt
│   ├── # 開発環境向けの依存パッケージをリストアップ。テストやコード解析ツールなどを含む
├── requirements.txt
│   ├── # プロジェクトの本番環境で必要な依存パッケージをリストアップ
├── source.bat
│   ├── # Windows環境での開発セットアップや環境変数の設定、依存関係のインストールを自動化するバッチスクリプト
└── tests
    ├── __init__.py
    │   ├── # Pythonパッケージとして認識させるための初期化ファイル
    └── unit
        ├── __init__.py
        │   ├── # Pythonパッケージとして認識させるための初期化ファイル
        └── test_cdk_crud_api_python_stack.py
            ├── # cdk_crud_api_python_stack.py内で定義されたスタックのユニットテストを実装

2. 仮想環境の作成と依存関係のインストール

次に、Pythonの仮想環境を作成し、必要なパッケージをインストールします。

python3 -m venv .venv
source .venv/bin/activate  # Windowsの場合は .\.venv\Scripts\activate
pip install --upgrade pip
pip install -r requirements.txt

requirements.txtに以下の内容を追加します。

aws-cdk-lib==2.xx
constructs>=10.0.0,<11.0.0
python-dotenv

さらに、必要なCDKモジュールをインストールします。

pip install aws-cdk-lib constructs python-dotenv

3. ディレクトリ構成の確認

初期化後のディレクトリ構成は以下のようになります。

cdk-crud-api-python/
├── README.md
├── app.py
├── cdk.json
├── cdk_crud_api_python
│   ├── __init__.py
│   └── cdk_crud_api_python_stack.py
├── requirements-dev.txt
├── requirements.txt
├── source.bat
└── tests
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── test_cdk_crud_api_python_stack.py

lambdaフォルダは後ほど作成しますが、今回はプレースホルダーとして表示しています。

CDKスタックの作成

次に、CDKスタックを定義して、DynamoDBテーブル、Lambda関数、API Gatewayを設定します。
DynamoDBのテーブル構成をカスタマイズしやすくするため、設定を外部化します。
また、IP制限などの設定も環境変数から読み取るようにします。

1. 必要なモジュールのインポート

cdk_crud_api_python/cdk_crud_api_python_stack.pyファイルを開き、以下のモジュールをインポートします。

from aws_cdk import (
    Stack,
    aws_dynamodb as dynamodb,
    aws_lambda as _lambda,
    aws_apigateway as apigateway,
    aws_iam as iam,
    RemovalPolicy
)
from constructs import Construct
import os
from dotenv import load_dotenv

load_dotenv()

2. 環境変数の設定

IPアドレスやDynamoDBのテーブル構成を環境変数から読み取るために、.envファイルを使用します。
プロジェクトルートに.envファイルを作成し、以下の内容を追加します。

ALLOWED_IPS=203.0.113.0/24,198.51.100.0/24
DYNAMODB_PARTITION_KEY=id
DYNAMODB_SORT_KEY= # 必要に応じて設定

.envファイルを読み込むために、Pythonのdotenvパッケージを使用します。
仮想環境をアクティベートした状態で、以下のコマンドを実行してインストールします。

pip install python-dotenv

その後、cdk_crud_api_python_stack.pyに以下を追加して環境変数を読み込みます。

from dotenv import load_dotenv

load_dotenv()

3. DynamoDBテーブルの定義

DynamoDBテーブルの構成を環境変数から読み取り、柔軟に定義できるようにします。

class CdkCrudApiPythonStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 環境変数から許可するIPアドレス範囲を取得
        allowed_ips = os.getenv("ALLOWED_IPS", "203.0.113.0/24,198.51.100.0/24").split(",")
        
        # DynamoDBのキー設定を環境変数から取得
        partition_key = os.getenv("DYNAMODB_PARTITION_KEY", "id")
        sort_key = os.getenv("DYNAMODB_SORT_KEY")  # 必要に応じて設定
        
        # DynamoDBテーブルの作成
        table = dynamodb.Table(
            self, "ItemsTable",
            partition_key=dynamodb.Attribute(
                name=partition_key,
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(name=sort_key, type=dynamodb.AttributeType.STRING) if sort_key else None,
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY  # 開発環境用。プロダクションでは注意
        )
        
        # 以下、Lambda関数やAPI Gatewayの設定に続く

ポイント解説:

環境変数の活用: DynamoDBのパーティションキーとソートキーを環境変数から取得し、柔軟にテーブル構成を変更可能にしています。ソートキーはオプションで設定できるようにしています。

4. Lambda関数の作成と設定

Lambda関数を作成し、環境変数としてDynamoDBのテーブル名と許可されたIPアドレスを設定します。

        # Lambda関数の作成
        crud_lambda = _lambda.Function(
            self, "CrudFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda"),
            environment={
                "TABLE_NAME": table.table_name,
                "ALLOWED_IPS": ",".join([ip.strip() for ip in allowed_ips])
            }
        )
        
        # LambdaにDynamoDBへのアクセス権限を付与
        table.grant_read_write_data(crud_lambda)

ポイント解説:

Lambda環境変数: DynamoDBのテーブル名と許可されたIPアドレス範囲を環境変数として設定することで、Lambda関数内で柔軟にこれらの情報を利用できます。

5. API Gatewayの設定とIP制限の適用

API Gatewayを設定し、Lambda関数との統合を行います。
また、IP制限をリソースポリシーとして適用します。

        # API Gatewayの作成
        api = apigateway.RestApi(
            self, "CrudApi",
            rest_api_name="CRUD Service",
            description="This service serves CRUD operations.",
            policy=iam.PolicyDocument(
                statements=[
                    iam.PolicyStatement(
                        effect=iam.Effect.ALLOW,
                        principals=[iam.AnyPrincipal()],
                        actions=["execute-api:Invoke"],
                        resources=["execute-api:/*"],
                        conditions={
                            "IpAddress": {"aws:SourceIp": allowed_ips}
                        }
                    )
                ]
            ),
            default_cors_preflight_options=apigateway.CorsOptions(
                allow_origins=apigateway.Cors.ALL_ORIGINS,
                allow_methods=apigateway.Cors.ALL_METHODS
            )
        )
        
        # APIリソースとメソッドの設定
        items = api.root.add_resource("items")
        items.add_method("GET", apigateway.LambdaIntegration(crud_lambda))
        items.add_method("POST", apigateway.LambdaIntegration(crud_lambda))
        
        single_item = items.add_resource("{id}")
        single_item.add_method("GET", apigateway.LambdaIntegration(crud_lambda))
        single_item.add_method("PUT", apigateway.LambdaIntegration(crud_lambda))
        single_item.add_method("DELETE", apigateway.LambdaIntegration(crud_lambda))

ポイント解説:

IP制限: PolicyDocumentを使用して、指定されたIPアドレス範囲からのみAPIへのアクセスを許可しています。環境変数から取得したIPアドレスを適用しています。これにより、add_to_resource_policyを使用せずに、API Gatewayの作成時に直接リソースポリシーを設定できます。
CORS設定: default_cors_preflight_optionsを設定することで、クロスオリジンリクエストを許可しています。今回はすべてのオリジンとメソッドを許可していますが、必要に応じて制限をかけることも可能です。
APIリソースとメソッド: /itemsと/items/{id}のエンドポイントを作成し、各HTTPメソッドにLambda関数を統合しています。

最終形

cdk_crud_api_python/cdk_crud_api_python_stack.py

from aws_cdk import (
    Stack,
    aws_dynamodb as dynamodb,
    aws_lambda as _lambda,
    aws_apigateway as apigateway,
    aws_iam as iam,
    RemovalPolicy
)
from constructs import Construct
import os
from dotenv import load_dotenv

load_dotenv()

class CdkCrudApiPythonStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 環境変数から許可するIPアドレス範囲を取得
        allowed_ips = os.getenv("ALLOWED_IPS", "203.0.113.0/24,198.51.100.0/24").split(",")
        
        # DynamoDBのキー設定を環境変数から取得
        partition_key = os.getenv("DYNAMODB_PARTITION_KEY", "id")
        sort_key = os.getenv("DYNAMODB_SORT_KEY")  # 必要に応じて設定
        
        # DynamoDBテーブルの作成
        table = dynamodb.Table(
            self, "ItemsTable",
            partition_key=dynamodb.Attribute(
                name=partition_key,
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(name=sort_key, type=dynamodb.AttributeType.STRING) if sort_key else None,
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY  # 開発環境用。プロダクションでは注意
        )
        
        # Lambda関数の作成
        crud_lambda = _lambda.Function(
            self, "CrudFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda"),
            environment={
                "TABLE_NAME": table.table_name,
                "ALLOWED_IPS": ",".join([ip.strip() for ip in allowed_ips])
            }
        )
        
        # LambdaにDynamoDBへのアクセス権限を付与
        table.grant_read_write_data(crud_lambda)
        
        # API Gatewayの作成
        api = apigateway.RestApi(
            self, "CrudApi",
            rest_api_name="CRUD Service",
            description="This service serves CRUD operations.",
            policy=iam.PolicyDocument(
                statements=[
                    iam.PolicyStatement(
                        effect=iam.Effect.ALLOW,
                        principals=[iam.AnyPrincipal()],
                        actions=["execute-api:Invoke"],
                        resources=["execute-api:/*"],
                        conditions={
                            "IpAddress": {"aws:SourceIp": allowed_ips}
                        }
                    )
                ]
            ),
            default_cors_preflight_options=apigateway.CorsOptions(
                allow_origins=apigateway.Cors.ALL_ORIGINS,
                allow_methods=apigateway.Cors.ALL_METHODS
            )
        )
        
        # APIリソースとメソッドの設定
        items = api.root.add_resource("items")
        items.add_method("GET", apigateway.LambdaIntegration(crud_lambda))
        items.add_method("POST", apigateway.LambdaIntegration(crud_lambda))
        
        single_item = items.add_resource("{id}")
        single_item.add_method("GET", apigateway.LambdaIntegration(crud_lambda))
        single_item.add_method("PUT", apigateway.LambdaIntegration(crud_lambda))
        single_item.add_method("DELETE", apigateway.LambdaIntegration(crud_lambda))

Lambda関数の実装

Lambda関数は、DynamoDBテーブルへのCRUD操作を行います。
また、IPアドレス制限もここで実装します。

1. Lambdaディレクトリの作成

プロジェクトのルートディレクトリにlambdaフォルダを作成し、その中にindex.pyファイルを作成します。

mkdir lambda
touch lambda/index.py

2. Lambdaハンドラーの実装

lambda/index.pyに以下のコードを追加します。
これは、柔軟なDynamoDBテーブル構成に対応し、IPアドレス制限を実装したLambdaハンドラーの例です。

import json
import os
import boto3
import uuid

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])

def handler(event, context):
    http_method = event['httpMethod']
    resource = event['resource']
    path_parameters = event.get('pathParameters', {})
    item_id = path_parameters.get('id') if path_parameters else None
    allowed_ips = os.environ.get('ALLOWED_IPS', '').split(',')
    
    # クライアントのIPアドレスを取得
    client_ip = event['requestContext']['identity']['sourceIp']
    if allowed_ips and not any(ip_match(client_ip, ip.strip()) for ip in allowed_ips):
        return {
            "statusCode": 403,
            "body": json.dumps({"message": "Forbidden"})
        }
    
    try:
        if resource == "/items" and http_method == "GET":
            # 全アイテムの取得
            response = table.scan()
            return {
                "statusCode": 200,
                "body": json.dumps(response.get('Items', []))
            }

        elif resource == "/items" and http_method == "POST":
            # 新しいアイテムの作成
            body = json.loads(event['body'])
            if 'id' not in body:
                body['id'] = generate_id()
            table.put_item(Item=body)
            return {
                "statusCode": 201,
                "body": json.dumps(body)
            }

        elif resource == "/items/{id}" and http_method == "GET":
            # 単一アイテムの取得
            response = table.get_item(Key={'id': item_id})
            item = response.get('Item')
            if item:
                return {
                    "statusCode": 200,
                    "body": json.dumps(item)
                }
            else:
                return {
                    "statusCode": 404,
                    "body": json.dumps({"message": "Item not found"})
                }

        elif resource == "/items/{id}" and http_method == "PUT":
            # アイテムの更新
            body = json.loads(event['body'])
            body['id'] = item_id
            table.put_item(Item=body)
            return {
                "statusCode": 200,
                "body": json.dumps(body)
            }

        elif resource == "/items/{id}" and http_method == "DELETE":
            # アイテムの削除
            table.delete_item(Key={'id': item_id})
            return {
                "statusCode": 204,
                "body": ""
            }

        else:
            return {
                "statusCode": 405,
                "body": json.dumps({"message": "Method Not Allowed"})
            }

    except Exception as e:
        print(e)
        return {
            "statusCode": 500,
            "body": json.dumps({"message": "Internal Server Error"})
        }

def generate_id():
    return str(uuid.uuid4())

def ip_match(client_ip, allowed_ip):
    """
    クライアントのIPアドレスが許可されたIPアドレス範囲にマッチするか確認します。
    CIDR表記をサポートします。
    """
    import ipaddress
    try:
        client = ipaddress.ip_address(client_ip)
        network = ipaddress.ip_network(allowed_ip, strict=False)
        return client in network
    except ValueError:
        return False

ポイント解説:

IPアドレス制限: 環境変数から取得したALLOWED_IPSをもとに、クライアントのIPアドレスが許可されているかをチェックしています。ipaddressモジュールを使用してCIDR表記のIPアドレス範囲に対応しています。
DynamoDBの柔軟な操作: パーティションキーやソートキーが柔軟に設定できるため、テーブル構成に応じたCRUD操作が可能です。
エラーハンドリング: 基本的なエラーハンドリングを実装しており、内部エラー時には500 Internal Server Errorを返します。

デプロイ準備とデプロイ

1. `app.py`の確認

app.pyファイルが以下のようになっていることを確認してください。
環境変数の設定もここで行います。

#!/usr/bin/env python3
import os
import aws_cdk as cdk

from cdk_crud_api_python.cdk_crud_api_python_stack import CdkCrudApiPythonStack

app = cdk.App()

CdkCrudApiPythonStack(app, "CdkCrudApiPythonStack",
                      env=cdk.Environment(
                          account=os.getenv('CDK_DEFAULT_ACCOUNT'),
                          region=os.getenv('CDK_DEFAULT_REGION')
                      )
)

app.synth()

2. CDKのブートストラップ（初回のみ）

CDKを初めてデプロイする場合は、ブートストラップを実行します。

cdk bootstrap

3. デプロイ

CDKスタックをデプロイします。

cdk deploy

デプロイが完了すると、API GatewayのエンドポイントURLが表示されます。
このURLを使用してCRUD操作を行うことができます。

最終的なディレクトリ構成

プロジェクト全体の最終的なディレクトリ構成は以下の通りです。

cdk-crud-api-python/
├── app.py
├── cdk_crud_api_python
│   ├── __init__.py
│   └── cdk_crud_api_python_stack.py
├── lambda
│   └── index.py
├── requirements.txt
├── .gitignore
├── README.md
├── cdk.json
└── .env

各ファイル・フォルダの役割:

app.py: CDKアプリケーションのエントリーポイント。
cdk_crud_api_python/: CDKスタックの定義が含まれるパッケージ。
- init.py: パッケージ初期化ファイル。
- cdk_crud_api_python_stack.py: DynamoDB、Lambda、API Gatewayのリソースを定義するスタックファイル。
lambda/: Lambda関数のコードが含まれるディレクトリ。
- index.py: CRUDロジックを実装したLambdaハンドラー。
requirements.txt: Pythonの依存関係リスト。
.gitignore: Gitで追跡しないファイル・フォルダを指定。
README.md: プロジェクトの説明や使用方法を記載。
cdk.json: CDKの設定ファイル。
.env: 環境変数を定義するファイル（セキュリティに注意）。

テスト方法

デプロイが完了したら、APIのエンドポイントを使用してCRUD操作をテストします。
以下は各操作の例です。

1. 全アイテムの取得 (`GET /items`)

curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items

2. 新しいアイテムの作成 (`POST /items`)

curl -X POST https://your-api-id.execute-api.region.amazonaws.com/prod/items \
     -H "Content-Type: application/json" \
     -d '{"name": "Item Name", "description": "Item Description"}'

3. 単一アイテムの取得 (`GET /items/{id}`)

curl -X GET https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id>

4. アイテムの更新 (`PUT /items/{id}`)

curl -X PUT https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id> \
     -H "Content-Type: application/json" \
     -d '{"name": "Updated Name", "description": "Updated Description"}'

5. アイテムの削除 (`DELETE /items/{id}`)

curl -X DELETE https://your-api-id.execute-api.region.amazonaws.com/prod/items/<item_id>

IP制限の確認:

許可されたIPアドレス範囲からのみ上記のリクエストが成功することを確認してください。
その他のIPからのアクセスは403 Forbiddenエラーとなります。

まとめ

今回のガイドでは、AWS CDK（Python）を使用して、DynamoDB、Lambda、およびAPI Gatewayを組み合わせた柔軟性の高いCRUD APIを構築し、IPアドレス制限を設定する方法を詳しく解説しました。
特に、DynamoDBのテーブル構成を環境変数から柔軟に設定できるようにすることで、カスタマイズ性を向上させています。

ポイントまとめ:

環境変数の活用: IPアドレスやDynamoDBのテーブル構成を環境変数から設定することで、柔軟な構成管理が可能になります。
DynamoDBの柔軟なテーブル構成: パーティションキーやソートキーを環境変数から取得することで、テーブル構成を簡単にカスタマイズできます。
IPアドレス制限: API Gatewayのリソースポリシーを使用して、指定されたIPアドレス範囲からのみAPIへのアクセスを許可します。
セキュリティの強化: 環境変数を適切に管理し、必要に応じて認証・認可を追加することで、セキュリティを強化できます。

このガイドを参考に、ぜひ自分のプロジェクトに応用してみてください。また、必要に応じてさらなる機能の追加やセキュリティの強化を行い、より堅牢なAPIを構築していきましょう！

参考資料:

会社紹介

株式会社 Mosaica
最先端テクノロジーで社会課題を解決し、持続可能な未来を創造する IT カンパニー。
AI ソリューション、クラウド統合、DX 推進、経営コンサルティングなど包括的なサービスでビジネス変革を支援しています。

詳しくは公式サイトまでお気軽にご相談ください。
公式サイト： https://mosaica.co.jp/