py#dynamodb

このジェネレーターは、Amazon DynamoDB をバックエンドとする新しい Python プロジェクトを作成します。エンティティモデリングには PynamoDB を使用します。AWS CDK または Terraform を使用して DynamoDB テーブルをプロビジョニングおよび管理するために必要なアプリケーションコードとインフラストラクチャを生成し、シングルテーブル設計のサポートと DynamoDB Local による組み込みのローカル開発環境を提供します。

使用方法

DynamoDB プロジェクトを生成する

VSCode
CLI

インストール Nx Console VSCode Plugin まだインストールしていない場合
VSCodeでNxコンソールを開く
クリック Generate (UI) "Common Nx Commands"セクションで
検索 @aws/nx-plugin - py#dynamodb
必須パラメータを入力
クリック Generate

pnpm nx g @aws/nx-plugin:py#dynamodb

yarn nx g @aws/nx-plugin:py#dynamodb

npx nx g @aws/nx-plugin:py#dynamodb

bunx nx g @aws/nx-plugin:py#dynamodb

変更されるファイルを確認するためにドライランを実行することもできます

pnpm nx g @aws/nx-plugin:py#dynamodb --dry-run

yarn nx g @aws/nx-plugin:py#dynamodb --dry-run

npx nx g @aws/nx-plugin:py#dynamodb --dry-run

bunx nx g @aws/nx-plugin:py#dynamodb --dry-run

オプション

パラメータ	型	デフォルト	説明
name 必須	string	-	生成するDynamoDBプロジェクトの名前
directory	string	packages	プロジェクトを格納するディレクトリ
subDirectory	string	-	プロジェクトが配置されるサブディレクトリ。デフォルトではプロジェクト名になります。
framework	pynamodb	pynamodb	DynamoDBエンティティに使用するフレームワーク
tableName	string	-	DynamoDBテーブル名。指定しない場合は自動生成されます。
infra	dynamodb \| none	dynamodb	DynamoDBテーブル用にプロビジョニングするインフラストラクチャ
iac	inherit \| cdk \| terraform	inherit	優先するIaCプロバイダー。デフォルトでは初期選択から継承されます。
preferInstallDependencies	boolean	true	ジェネレーター実行後に依存関係のインストールを優先するかどうか。複数のジェネレーターをバッチ処理する際にインストールを延期する場合はfalseに設定します（後続のジェネレーターがNxプロジェクトグラフを計算できるよう、必要に応じてインストールは実行されます）。最後に一度だけインストールします。

ジェネレーターの出力

ジェネレーターは <directory>/<name> ディレクトリに以下のプロジェクト構造を作成します：

Directory<name>
- __init__.py パッケージのエクスポート
- client.py DynamoDB クライアントとテーブル名の解決
- Directoryentities
  - base.py GSI 宣言を含むベース PynamoDB モデル
  - example.py エンティティ定義の例
  - __init__.py エンティティのエクスポート
config.json GSI 定義とローカル開発設定を含むテーブル設定
project.json プロジェクト設定とビルドターゲット

ローカル開発スクリプトはすべての DynamoDB プロジェクト（TypeScript と Python の両方）間で共有され、一度だけ以下の場所に生成されます：

Directorypackages/common/scripts/src/dynamodb
- create-local-table.ts ローカル DynamoDB Local インスタンスに DynamoDB テーブルを作成する
- pull-image.ts DynamoDB Local イメージをプルする
- start-container.ts DynamoDB Local コンテナを起動する

インフラストラクチャ

このジェネレータは選択した iacProvider に基づいてInfrastructure as Codeを生成するため、packages/common に関連するCDKコンストラクトまたはTerraformモジュールを含むプロジェクトを作成します。

共通のInfrastructure as Codeプロジェクトは以下の構造を持ちます：

CDK
Terraform

Directorypackages/common/constructs
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用コンストラクト
    …
  - Directorycore/ app 内のコンストラクトで再利用される汎用コンストラクト
    …
  - index.ts app からコンストラクトをエクスポートするエントリーポイント
- project.json プロジェクトのビルドターゲットと設定

Directorypackages/common/terraform
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用Terraformモジュール
    …
  - Directorycore/ app 内のモジュールで再利用される汎用モジュール
    …
- project.json プロジェクトのビルドターゲットと設定

CDK
Terraform

Directorypackages/common/constructs/src
- Directoryapp
  - Directorydynamodb
    <name>.ts テーブル固有のインフラストラクチャ
- Directorycore
  - dynamodb.ts 汎用 DynamoDB テーブルコンストラクト

ローカル開発

ローカル DynamoDB の起動

ジェネレーターは、DynamoDB Local インスタンスを起動してテーブルを作成する dev ターゲットを設定します。プロジェクトの dev ターゲットを使用します：

pnpm nx dev <project-name>

yarn nx dev <project-name>

npx nx dev <project-name>

bunx nx dev <project-name>

これにより自動的に以下が実行されます：

DynamoDB Local イメージをプル（pull-image ターゲット）
コンテナを起動
config.json で定義されたインデックスを持つローカルテーブルを作成

データモデリング

生成されたプロジェクトは、エンティティモデリングに PynamoDB を使用します。すべてのエンティティは生成された BaseModel を継承しなければなりません — これは実行時に正しい DynamoDB テーブル名を解決し、デプロイ時には AWS AppConfig から、DynamoDB Local 経由でローカル実行時には config.json から読み取ります。これがないと、PynamoDB はどのテーブルを使用すればよいかわかりません。BaseModel は PynamoDB のポリモーフィズムサポートも使用して、複数のエンティティタイプを単一のテーブルに格納します。これは DynamoDB のシングルテーブル設計に従っています。

生成されたサンプルエンティティを出発点として、<name>/entities/ 配下にエンティティファイルを追加または更新します：

from collections.abc import Iterator
from datetime import UTC, datetime
from pynamodb.attributes import UnicodeAttribute
from .base import BaseModel


class ExampleModel(BaseModel, discriminator='ExampleModel'):
    """
    キー設計:
      pk=EXAMPLE#<id>,        sk=EXAMPLE#<id>
      gsi1pk=CATEGORY#<cat>,  gsi1sk=EXAMPLE#<id>    <- カテゴリ別にアイテムをリスト
      gsi2pk=EXAMPLE,         gsi2sk=<created_at>    <- 日付別にすべてのアイテムをリスト
    """

    name = UnicodeAttribute()
    category = UnicodeAttribute()
    created_at = UnicodeAttribute()
    updated_at = UnicodeAttribute()

    @classmethod
    def make_pk(cls, id: str) -> str:
        return f'EXAMPLE#{id}'

    @classmethod
    def create(cls, id: str, name: str, category: str) -> 'ExampleModel':
        now = datetime.now(UTC).isoformat()
        item = cls(
            pk=cls.make_pk(id),
            sk=cls.make_pk(id),
            gsi1pk=f'CATEGORY#{category}',
            gsi1sk=cls.make_pk(id),
            gsi2pk='EXAMPLE',
            gsi2sk=now,
            name=name,
            category=category,
            created_at=now,
            updated_at=now,
        )
        item.save()
        return item

    # ── Primary index ─────────────────────────────────────────────────────────
    @classmethod
    def get_by_id(cls, id: str) -> 'ExampleModel':
        return cls.get(cls.make_pk(id), cls.make_pk(id))

    # ── gsi1_index: partition=category, sort=id ───────────────────────────────
    @classmethod
    def list_by_category(cls, category: str) -> Iterator['ExampleModel']:
        return cls.gsi1_index.query(f'CATEGORY#{category}')

    # ── gsi2_index: partition=type, sort=created_at ───────────────────────────
    @classmethod
    def list_created_between(cls, start: datetime, end: datetime) -> Iterator['ExampleModel']:
        return cls.gsi2_index.query(
            'EXAMPLE',
            range_key_condition=ExampleModel.gsi2sk.between(
                start.isoformat(), end.isoformat(),
            ),
            scan_index_forward=False,
        )

詳細については、PynamoDB チュートリアルを参照してください。

アクセスパターンを中心とした設計

DynamoDB では、スキーマ設計はデータの形状ではなく、クエリから始まります。モデルを書く前に、アプリケーションが必要とするすべてのアクセスパターンをリストアップし、各パターンが単一のテーブルリクエストで応答できるように pk、sk、GSI キー値を設計します — JOIN も順次読み取りも不要です。

生成された ExampleModel は、3 つのパターンについてこれを実証しています：

ID で取得 — プライマリインデックス、pk=EXAMPLE#<id>、sk=EXAMPLE#<id>
カテゴリ別にリスト — gsi1、pk=CATEGORY#<category>
作成日別にリスト — gsi2、pk=EXAMPLE、ISO タイムスタンプ間のソートキー

タイプ接頭辞の規約（例：EXAMPLE#、CATEGORY#）は意図的なものです：テーブルを閲覧する際にアイテムを自己記述的にし、インデックスを共有するエンティティタイプ間の偶発的なキー衝突を防ぎ、begins_with を使用したソートキー接頭辞フィルタリングを可能にします。

新しいエンティティを書く前に、そのキーパターンを docstring で事前に定義してください。次のセクションの OrderModel はこの規約に従っています：

class OrderModel(BaseModel, discriminator='OrderModel'):
    """
    キー設計:
      pk=ORDER#<order_id>,    sk=ORDER#<order_id>
      gsi1pk=USER#<user_id>,  gsi1sk=ORDER#<order_id>   <- ユーザーの注文をリスト
      gsi2pk=ORDER,           gsi2sk=<created_at>        <- 日付別にすべての注文をリスト
    """

複数のエンティティタイプの格納

PynamoDB の DiscriminatorAttribute は、すべてのアイテムにタイプラベル（entity_type）を格納します。BaseModel 経由でクエリする場合、このラベルは各結果を正しいサブクラスとして自動的にインスタンス化するために使用されます — したがって、単一のクエリで UserModel、OrderModel、および同じテーブルに登録されている他のエンティティタイプの混合を返すことができます。

PynamoDB の DiscriminatorAttribute は、単一の同質なテーブル内のクラス階層ポリモーフィズム用に設計されました — 構造的に異なるエンティティタイプを 1 つのテーブルに格納するためではありません。ここで使用されているパターンは機能しますが、これらの注意点を念頭に置いてください：

キー衝突はあなたの責任です。 PynamoDB は、2 つのエンティティタイプが同じ pk/sk ペアを書き込むことを防ぎません。常に一意のタイプ接頭辞（例：ORDER#、USER#）を使用してください。
エンティティ間クエリには BaseModel が必要です。 PynamoDB は特定のサブクラスをクエリする際に自動的に識別子フィルタを追加するため、同じキーにある他のエンティティタイプのアイテムは決して返されません。共有 GSI パーティションキーからタイプの混合を取得するには、常に BaseModel 経由でクエリしてください。
共有テーブル設定。 すべてのエンティティは、BaseModel で定義された Meta 設定（テーブル名、リージョン、認証情報）を共有します。

以下は、同じテーブルに格納された UserModel と関連する OrderModel レコードを持つ、完全な 2 エンティティの例です：

from collections.abc import Iterator
from datetime import UTC, datetime
from pynamodb.attributes import UnicodeAttribute
from .base import BaseModel


class UserModel(BaseModel, discriminator='UserModel'):
    """
    キー設計:
      pk=USER#<user_id>,  sk=USER#<user_id>
      gsi2pk=USER,        gsi2sk=<created_at>    <- 日付別にすべてのユーザーをリスト
    """

    username = UnicodeAttribute()
    email = UnicodeAttribute()
    created_at = UnicodeAttribute()

    @classmethod
    def make_pk(cls, user_id: str) -> str:
        return f'USER#{user_id}'

    @classmethod
    def create(cls, user_id: str, username: str, email: str) -> 'UserModel':
        now = datetime.now(UTC).isoformat()
        item = cls(
            pk=cls.make_pk(user_id),
            sk=cls.make_pk(user_id),
            gsi2pk='USER',
            gsi2sk=now,
            username=username,
            email=email,
            created_at=now,
        )
        item.save()
        return item

    @classmethod
    def get_by_id(cls, user_id: str) -> 'UserModel':
        return cls.get(cls.make_pk(user_id), cls.make_pk(user_id))

    @classmethod
    def list_recent(cls, limit: int | None = None) -> Iterator['UserModel']:
        return cls.gsi2_index.query('USER', limit=limit, scan_index_forward=False)

from collections.abc import Iterator
from datetime import UTC, datetime
from pynamodb.attributes import UnicodeAttribute
from .base import BaseModel


class OrderModel(BaseModel, discriminator='OrderModel'):
    """
    キー設計:
      pk=ORDER#<order_id>,    sk=ORDER#<order_id>
      gsi1pk=USER#<user_id>,  gsi1sk=ORDER#<order_id>    <- ユーザー別に注文をリスト
      gsi2pk=ORDER,           gsi2sk=<created_at>         <- 日付別にすべての注文をリスト
    """

    user_id = UnicodeAttribute()
    total = UnicodeAttribute()
    created_at = UnicodeAttribute()

    @classmethod
    def make_pk(cls, order_id: str) -> str:
        return f'ORDER#{order_id}'

    @classmethod
    def create(cls, order_id: str, user_id: str, total: str) -> 'OrderModel':
        now = datetime.now(UTC).isoformat()
        item = cls(
            pk=cls.make_pk(order_id),
            sk=cls.make_pk(order_id),
            gsi1pk=f'USER#{user_id}',
            gsi1sk=cls.make_pk(order_id),
            gsi2pk='ORDER',
            gsi2sk=now,
            user_id=user_id,
            total=total,
            created_at=now,
        )
        item.save()
        return item

    @classmethod
    def get_by_id(cls, order_id: str) -> 'OrderModel':
        return cls.get(cls.make_pk(order_id), cls.make_pk(order_id))

    # ── gsi1_index: partition=user, sort=order_id ────────────────────────────
    @classmethod
    def list_by_user(cls, user_id: str) -> Iterator['OrderModel']:
        return cls.gsi1_index.query(f'USER#{user_id}')

    # ── gsi2_index: partition=type, sort=created_at ───────────────────────────
    @classmethod
    def list_recent(cls, limit: int | None = None) -> Iterator['OrderModel']:
        return cls.gsi2_index.query('ORDER', limit=limit, scan_index_forward=False)

__init__.py から新しいエンティティをエクスポートします：

from .user import UserModel
from .order import OrderModel
from .example import ExampleModel

GSI オーバーロード

BaseModel は 2 つの共有 GSI（gsi1_index、gsi2_index）を提供します。上記の UserModel と OrderModel の両方が gsi2 に書き込みますが、異なる gsi2pk 値（USER vs ORDER）を使用します。これが GSI オーバーロードです：単一の物理インデックスを再利用して、追加の GSI 容量を消費することなく、複数の独立したアクセスパターンを提供します。

UserModel — gsi2pk=USER、gsi2sk=<created_at> → 日付別にすべてのユーザーをリスト
OrderModel — gsi2pk=ORDER、gsi2sk=<created_at> → 日付別にすべての注文をリスト

gsi1 も、複数のエンティティタイプが同じ親を共有する場合にオーバーロードできます。後で同じくユーザーに属する ReviewModel を追加する場合、REVIEW#<id> ソートキーで gsi1pk=USER#<user_id> を割り当てることができます — 追加の GSI は不要です。BaseModel 経由で gsi1 をクエリすると、そのユーザーの注文とレビューの両方が 1 つのリクエストで返され、PynamoDB が各アイテムを正しいサブクラスとしてインスタンス化します：

from .base import BaseModel
from .order import OrderModel
from .review import ReviewModel

user_id = 'user-123'
items = list(BaseModel.gsi1_index.query(f'USER#{user_id}'))

orders  = [i for i in items if isinstance(i, OrderModel)]
reviews = [i for i in items if isinstance(i, ReviewModel)]

オーバーロードされた GSI から 1 つのエンティティタイプのみを取得するには、ソートキー接頭辞条件を使用します：

orders_only = list(BaseModel.gsi1_index.query(
    f'USER#{user_id}',
    range_key_condition=BaseModel.gsi1sk.startswith('ORDER#'),
))

1 対多の関係

1 対多の関係では、子エンティティが GSI パーティションキーに親への参照を格納し、データを複製することなく両方向に関係をトラバース可能にします。上記の UserModel / OrderModel の例はまさにこのパターンです：

ID で単一の注文を取得 — プライマリテーブル：pk=ORDER#<id>、sk=ORDER#<id>
ユーザーのすべての注文をリスト — gsi1：pk=USER#<user_id>

GSI ベースのルックアップの代替は、アイテムコレクションパターンです：子アイテムに親と同じ pk を与え、ソートキーを使用してそれらを区別します。これにより、GSI なしで、単一のプライマリテーブルクエリで親とそのすべての子を取得できます：

class OrderModel(BaseModel, discriminator='OrderModel'):
    """
    キー設計（アイテムコレクション）:
      pk=USER#<user_id>,    sk=ORDER#<order_id>   <- 親ユーザーの下に共存
    """
    ...

# 単一のプライマリテーブルクエリでユーザーとそのすべての注文を取得
# BaseModel は DiscriminatorAttribute を介して各アイテムを正しいサブクラスにディスパッチ
items = list(BaseModel.query(f'USER#{user_id}'))
user   = next(i for i in items if isinstance(i, UserModel))
orders = [i for i in items if isinstance(i, OrderModel)]

トレードオフ：アイテムコレクションはすべての子を単一のパーティションキーの下に配置します。これはほとんどのワークロードに最適ですが、極端な書き込みスループットではホットパーティションを作成する可能性があります。GSI アプローチ（上記の例で使用）は、各エンティティを独自のパーティションに保持し、一般的に開始するのがより安全です。

多対多の関係

多対多の関係には、隣接リストパターンを使用したジャンクションエンティティが必要です：各リンクを記録する専用アイテムで、その GSI キーが方向を反転させて、関係を両方向にトラバース可能にします。

ArticleModel と TagModel を考えてみましょう。記事は多くのタグを持つことができ、タグは多くの記事に適用できます：

from collections.abc import Iterator
from pynamodb.attributes import UnicodeAttribute
from .base import BaseModel


class ArticleTagModel(BaseModel, discriminator='ArticleTag'):
    """
    Article ↔ Tag 多対多関係のジャンクションエンティティ。

    キー設計:
      pk=ARTICLE#<article_id>,  sk=TAG#<tag_name>             <- 記事のタグをリスト
      gsi1pk=TAG#<tag_name>,    gsi1sk=ARTICLE#<article_id>  <- タグの記事をリスト
    """

    article_id = UnicodeAttribute()
    tag_name = UnicodeAttribute()

    @classmethod
    def add(cls, article_id: str, tag_name: str) -> 'ArticleTagModel':
        item = cls(
            pk=f'ARTICLE#{article_id}',
            sk=f'TAG#{tag_name}',
            gsi1pk=f'TAG#{tag_name}',
            gsi1sk=f'ARTICLE#{article_id}',
            article_id=article_id,
            tag_name=tag_name,
        )
        item.save()
        return item

    @classmethod
    def remove(cls, article_id: str, tag_name: str) -> None:
        cls.get(f'ARTICLE#{article_id}', f'TAG#{tag_name}').delete()

    # ── Primary index: pk=article, sk=tag ─────────────────────────────────────
    @classmethod
    def list_tags_for_article(cls, article_id: str) -> Iterator['ArticleTagModel']:
        return cls.query(f'ARTICLE#{article_id}')

    # ── gsi1_index: pk=tag, sk=article ────────────────────────────────────────
    @classmethod
    def list_articles_for_tag(cls, tag_name: str) -> Iterator['ArticleTagModel']:
        return cls.gsi1_index.query(f'TAG#{tag_name}')

ArticleTagModel は pk=ARTICLE#<article_id> を使用します — 記事自体と同じパーティション — したがって、単一のプライマリテーブルクエリで記事とそのすべてのタグを取得できます：

from .base import BaseModel
from .article import ArticleModel
from .article_tag import ArticleTagModel

items   = list(BaseModel.query('ARTICLE#article-123'))
article = next(i for i in items if isinstance(i, ArticleModel))
tags    = [i.tag_name for i in items if isinstance(i, ArticleTagModel)]

DynamoDB データモデリングの詳細については、DynamoDB データモデリングガイドおよび Amazon DynamoDB でシングルテーブル設計を作成するを参照してください。

DynamoDB クライアントの使用

生成された client.py は 2 つの主要なユーティリティをエクスポートします：

is_local() — LOCAL_DEV=true の場合に True を返し、ローカルと AWS の動作を切り替えるために使用されます。
get_table_name() — DynamoDB テーブル名を返します。LOCAL_DEV=true の場合、config.json の localDev.tableName からテーブル名を読み取ります。それ以外の場合は、RUNTIME_CONFIG_APP_ID 環境変数を使用して AWS AppConfig から名前を取得し、後続の呼び出しのためにキャッシュします。

entities/base.py の BaseModel は両方を使用して PynamoDB を自動的に設定します：

接続 — BaseModel.Meta は、is_local() が True の場合、config.json から host を設定し、region、aws_access_key_id、aws_secret_access_key をハードコードして、PynamoDB をローカル DynamoDB インスタンスに向けます。AWS では、これらは未設定のままにされ、PynamoDB はデフォルトの認証情報チェーンを使用します。
テーブル名 — BaseModel._get_connection() は各操作の前に get_table_name() を呼び出すため、手動設定なしで実行時に正しいテーブルが解決されます。

ローカル DynamoDB の停止

dev を停止する（例：Ctrl+C で）と、DynamoDB Local コンテナは自動的に削除されますが、名前付きボリュームは保持されるため、再起動後もデータは永続化されます。

Windows でのシグナル処理の制限により、dev を停止してもコンテナは自動的に削除されません。手動で削除する必要があります：

<engine> rm -f <scope>-dynamodb

<engine> をコンテナエンジン（docker または finch）に、<scope> を Nx ワークスペースのスコープ（例：proj）に置き換えてください。

グローバルセカンダリインデックスの追加/削除

GSI は、プロジェクトルートの config.json の tableConfig.globalSecondaryIndexes キーで定義されます。各 GSI のエントリを追加し、<name>/entities/base.py の BaseModel に対応する GlobalSecondaryIndex クラスと属性を追加または削除して変更を反映します：

{
  ...
  "tableConfig": {
    "globalSecondaryIndexes": [
      {
        "indexName": "gsi1pk-gsi1sk-index",
        "partitionKey": "gsi1pk",
        "sortKey": "gsi1sk"
      },
      {
        "indexName": "gsi2pk-gsi2sk-index",
        "partitionKey": "gsi2pk",
        "sortKey": "gsi2sk"
      }
    ]
  }
}

sortKeyフィールドは、ハッシュキーのみのGSIの場合はオプションです。

この設定ファイルは、すべての利用者が読み取る唯一の信頼できる情報源です：

ローカル開発 — devはconfig.jsonを読み取り、GSIリストに一致するようにローカルテーブルを作成または更新します
CDK — コンストラクトは合成時にconfig.jsonを読み取るため、GSIの変更は次回のcdk deployに反映されます
Terraform — モジュールはplan/apply時にconfig.jsonを読み取ります

デプロイごとに1つのGSI

テーブルへの接続

任意の Python プロジェクトで、DynamoDB パッケージをワークスペースの依存関係として追加し、エンティティクラスを直接インポートします：

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

テーブルのデプロイ

DynamoDBジェネレーターは、選択したiacに基づいてCDKまたはTerraformインフラストラクチャを作成します。

CDK
Terraform

CDKコンストラクトはcommon/constructsに作成されます。使用例：

import { MyTable } from ':my-scope/common-constructs';

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

    const table = new MyTable(this, 'Table');
  }
}

これにより、以下の設定でDynamoDBテーブルがプロビジョニングされます：

pk（パーティションキー）とsk（ソートキー）、両方ともString型
config.jsonで定義されたグローバルセカンダリインデックス
オンデマンド（PAY_PER_REQUEST）課金
自動キーローテーション付きのカスタマー管理KMS暗号化
ポイントインタイムリカバリが有効
削除保護が有効
AWS AppConfigのdynamodb名前空間下のランタイム設定にテーブル名が登録

Terraformモジュールはcommon/terraformに作成されます。使用例：

module "my_table" {
  source = "../../common/terraform/src/app/dynamodb/my-table"
}

これにより、以下の設定でDynamoDBテーブルがプロビジョニングされます：

pk（パーティションキー）とsk（ソートキー）、両方ともString型
config.jsonで定義されたグローバルセカンダリインデックス
オンデマンド（PAY_PER_REQUEST）課金
自動キーローテーション付きのカスタマー管理KMS暗号化
ポイントインタイムリカバリが有効
削除保護が有効
ランタイム設定にテーブル名が登録

アクセス権限の付与

Lambda 関数が DynamoDB テーブルにアクセスできるようにするには、インフラストラクチャで必要な権限を付与します。

CDK
Terraform

テーブルコンストラクトで grantReadWriteData を呼び出します。これにより、Lambda 実行ロールに必要な DynamoDB と KMS の両方の権限が付与されます：

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table');

const api = new Api(this, 'Api', {
  integrations: Api.defaultIntegrations(this).build(),
});

Object.entries(api.integrations).forEach(([, integration]) => {
  table.grantReadWriteData(integration.handler);
});

Lambda 実行ロールに DynamoDB テーブルとその KMS 暗号化キーへのアクセス権限を付与します：

module "my_table" {
  source = "../../common/terraform/src/app/dynamodb/my-table"
}

resource "aws_iam_role_policy" "dynamodb_access" {
  role = module.my_api.lambda_role_name

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "dynamodb:GetItem",
          "dynamodb:PutItem",
          "dynamodb:UpdateItem",
          "dynamodb:DeleteItem",
          "dynamodb:Query",
          "dynamodb:Scan",
          "dynamodb:BatchGetItem",
          "dynamodb:BatchWriteItem",
        ]
        Resource = [
          module.my_table.table_arn,
          "${module.my_table.table_arn}/index/*",
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "kms:Encrypt",
          "kms:Decrypt",
          "kms:ReEncrypt*",
          "kms:GenerateDataKey*",
          "kms:DescribeKey"
        ]
        Resource = [module.my_table.kms_key_arn]
      },
    ]
  })
}

削除保護

削除保護はデフォルトで有効になっており、誤ってテーブルを削除することを防ぎます。

削除保護の無効化

短期間の開発環境やプレビュースタックなど、テーブルの削除が想定される環境では無効にします。

CDK
Terraform

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  deletionProtection: false,
});

module "my_table" {
  source                      = "../../common/terraform/src/app/dynamodb/my-table"
  deletion_protection_enabled = false
}

課金モード

テーブルはデフォルトでオンデマンド（PAY_PER_REQUEST）課金になります。予測可能で高スループットのワークロードには、プロビジョニングされたキャパシティに切り替えます。

CDK
Terraform

import { BillingMode } from 'aws-cdk-lib/aws-dynamodb';
import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  billingMode: BillingMode.PROVISIONED,
  readCapacity: 5,
  writeCapacity: 5,
});

module "my_table" {
  source       = "../../common/terraform/src/app/dynamodb/my-table"
  billing_mode = "PROVISIONED"
}

ポイントインタイムリカバリ

ポイントインタイムリカバリはデフォルトで有効になっており、過去35日間の任意の時点にテーブルを復元できます。

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  pointInTimeRecoverySpecification: { pointInTimeRecoveryEnabled: false },
});

module "my_table" {
  source                         = "../../common/terraform/src/app/dynamodb/my-table"
  point_in_time_recovery_enabled = false
}

暗号化キーのローテーション

テーブルの暗号化に使用されるKMSキーは、デフォルトで自動キーローテーションが有効になっています。セキュリティポリシーで外部的にローテーションを管理している場合は無効にします。

暗号化キーのローテーションの無効化

CDK
Terraform

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  enableKeyRotation: false,
});

module "my_table" {
  source              = "../../common/terraform/src/app/dynamodb/my-table"
  enable_key_rotation = false
}

接続

connectionジェネレータを使用して、このプロジェクトをワークスペース内の他のプロジェクトと統合できます。このプロジェクトに関連する接続は以下の通りです：

FastAPI to Python DynamoDB FastAPIをDynamoDBテーブルに接続する

Python Agent to Python DynamoDB Python AgentをDynamoDBテーブルに接続する

Python MCP Server to Python DynamoDB Python MCP ServerをDynamoDBテーブルに接続する