Python DynamoDB

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

使用方法

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

このジェネレーターを実行@aws/nx-plugin:py#dynamodb

CLI
VSCode

$ 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

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

コマンドを組み立てる8

name必須

オプション

ジェネレーターオプション8 オプション

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

ジェネレーターの出力

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

Directory<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
- Directoryentities
  - base.py Base PynamoDB model with GSI declarations
  - example.py Example entity definition
  - __init__.py Entity exports
config.json Table configuration including GSI definitions and local development settings
project.json Project configuration and build targets

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

Directorypackages/common/scripts/src/dynamodb
- create-local-table.ts Creates the DynamoDB table in the local DynamoDB Local instance
- pull-image.ts Pulls the DynamoDB Local image
- start-container.ts Starts the DynamoDB Local container

インフラストラクチャ

このジェネレーターは、選択した iac に基づいてインフラストラクチャをコードとして提供するため、関連する CDK コンストラクトまたは Terraform モジュールを含む packages/common にプロジェクトを作成します。

共通のインフラストラクチャコードプロジェクトは、次のように構成されています：

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 テーブルコンストラクト

アーキテクチャ

デプロイされたプロジェクトはテーブル自体をプロビジョニングし、接続されたすべてのプロジェクトがそれを読み書きします：

Loading the diagram…

ローカル開発

ローカル 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'):
    """
    Key design:
      pk=EXAMPLE#<id>,        sk=EXAMPLE#<id>
      gsi1pk=CATEGORY#<cat>,  gsi1sk=EXAMPLE#<id>    <- list items by category
      gsi2pk=EXAMPLE,         gsi2sk=<created_at>    <- list all items by date
    """

    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'):
    """
    Key design:
      pk=ORDER#<order_id>,    sk=ORDER#<order_id>
      gsi1pk=USER#<user_id>,  gsi1sk=ORDER#<order_id>   <- list orders for a user
      gsi2pk=ORDER,           gsi2sk=<created_at>        <- list all orders by date
    """

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

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'):
    """
    Key design:
      pk=USER#<user_id>,  sk=USER#<user_id>
      gsi2pk=USER,        gsi2sk=<created_at>    <- list all users by date
    """

    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'):
    """
    Key design:
      pk=ORDER#<order_id>,    sk=ORDER#<order_id>
      gsi1pk=USER#<user_id>,  gsi1sk=ORDER#<order_id>    <- list orders by user
      gsi2pk=ORDER,           gsi2sk=<created_at>         <- list all orders by date
    """

    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 対 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#'),
))

一対多リレーションシップ

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

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

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

class OrderModel(BaseModel, discriminator='OrderModel'):
    """
    Key design (item collection):
      pk=USER#<user_id>,    sk=ORDER#<order_id>   <- co-located under the parent user
    """
    ...

# Retrieve the user and all their orders in one primary-table query
# BaseModel dispatches each item to its correct subclass via 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'):
    """
    Junction entity for the Article ↔ Tag many-to-many relationship.

    Key design:
      pk=ARTICLE#<article_id>,  sk=TAG#<tag_name>             <- list tags for an article
      gsi1pk=TAG#<tag_name>,    gsi1sk=ARTICLE#<article_id>  <- list articles for a tag
    """

    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')

舞台裏では、ExampleModel は get_table_name() を呼び出して、実行時に AWS AppConfig からテーブル名を取得します。

テーブルのデプロイ

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 名前空間下の Runtime Config にテーブル名が登録される

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

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

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

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

core/runtime-config/appconfig モジュールはデフォルトで dynamodb 名前空間を公開するため、テーブル名は追加の設定なしでデプロイされます。そのモジュールに明示的に namespaces を渡す場合は、リストに dynamodb を含めてください。そうしないと、設定プロファイルが作成されず、生成されたテーブルクライアントがテーブル名を解決できなくなります。

削除保護

テーブルは 2 つの独立したガードによって保護されているため、どちらか一方だけをオフにしてもデータを削除することはできません：

CDK
Terraform

deletionProtection、DynamoDB によって強制されます。
RemovalPolicy.RETAIN、CloudFormation によって強制され、スタックから削除されてもテーブルをそのまま残します。

deletion_protection_enabled、DynamoDB によって強制されます。
common/terraform/src/core/dynamodb/dynamodb.tf のテーブルに対する lifecycle { prevent_destroy = true }、Terraform によって強制され、テーブルを破棄するプランは失敗します。

テーブルの削除

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

CDK
Terraform

import { RemovalPolicy } from 'aws-cdk-lib';
import { MyTable } from '@my-scope/common-constructs';

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

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

prevent_destroy はリテラルでなければなりません — Terraform は変数を参照することを許可していません — そのため、main.tf からオフにすることはできません。また、common/terraform/src/core/dynamodb/dynamodb.tf のテーブルから lifecycle ブロックを削除してください：

resource "aws_dynamodb_table" "table" {
  # ...

  lifecycle {
    prevent_destroy = true
  }
}

課金モード

テーブルはデフォルトでオンデマンド（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 キーで暗号化され、自動的に作成されます。暗号化を異なる方法で管理する場合は、AWS 管理キー、AWS 所有キー、または独自の KMS キーに切り替えます。

AWS 管理キーを使用する

AWS があなたに代わって管理する共有 aws/dynamodb KMS キーを使用します。アカウントの KMS コンソールに表示され、リクエストごとに課金されますが、作成、ローテーション、削除するキーはありません。

CDK
Terraform

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

const table = new MyTable(this, 'Table', {
  encryption: TableEncryption.AWS_MANAGED,
});

AWS_MANAGED を選択すると、テーブルはカスタマー管理キーで暗号化されなくなり、Checkov の CKV_AWS_119 に失敗します。テーブルで抑制してください：

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table, ['CKV_AWS_119'], 'Using an AWS managed key rather than a customer-managed CMK');

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

AWS 所有キーを使用する

AWS が完全に所有および管理するキーを使用します — 無料で、アカウントにキーが表示されることはありません。コンプライアンス上の理由でカスタマーまたはアカウントに表示されるキーが不要な場合の最もシンプルなオプションです。

CDK
Terraform

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

const table = new MyTable(this, 'Table', {
  encryption: TableEncryption.DEFAULT,
});

DEFAULT を選択すると、テーブルはカスタマー管理キーで暗号化されなくなり、Checkov の CKV_AWS_119 に失敗します。テーブルで抑制してください：

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table, ['CKV_AWS_119'], 'Using the AWS owned key rather than a customer-managed CMK');

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

CUSTOMER_MANAGED からの切り替え

すでにデプロイされているテーブルで、encryption を CUSTOMER_MANAGED から（AWS_MANAGED または DEFAULT のいずれかに）単一の terraform apply で変更すると失敗します：Terraform はテーブルを更新する前にカスタマー管理キーを破棄し、DynamoDB はキーがすでに削除保留中であるため更新を拒否します。

回避策として、まず AWS CLI を介してテーブルの暗号化を直接更新し、その後 Terraform に追いつかせて孤立したキーをクリーンアップします：

# For AWS_MANAGED:
aws dynamodb update-table --table-name <table-name> \
  --sse-specification Enabled=true,SSEType=KMS,KMSMasterKeyId=alias/aws/dynamodb

# For DEFAULT:
aws dynamodb update-table --table-name <table-name> --sse-specification Enabled=false

# Then wait for this to report ENABLED (or for SSEDescription to disappear, for DEFAULT):
aws dynamodb describe-table --table-name <table-name> --query Table.SSEDescription.Status

次に、Terraform 設定で encryption を更新し、通常通り terraform apply を実行します — Terraform は、何も依存していない、すでに使用されていないキーを破棄するだけで済みます。

独自の KMS キーを使用する

自動作成されるキーの代わりに、既存のカスタマー管理キーを提供します。キーは、独自のキーポリシーで DynamoDB サービスに必要な権限をすでに付与している必要があります。

CDK
Terraform

import { Key } from 'aws-cdk-lib/aws-kms';
import { MyTable } from '@my-scope/common-constructs';

const key = Key.fromKeyArn(this, 'Key', 'arn:aws:kms:us-east-1:111111111111:key/my-key-id');

const table = new MyTable(this, 'Table', {
  encryptionKey: key,
});

module "my_table" {
  source      = "../../common/terraform/src/app/dynamodb/my-table"
  kms_key_arn = "arn:aws:kms:us-east-1:111111111111:key/my-key-id"
}

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

テーブルが独自のカスタマー管理 KMS キーを作成する場合（デフォルトで、独自のキーを提供していない場合のみ）、そのキーは自動キーローテーションがデフォルトで有効になっています。セキュリティポリシーで外部的にローテーションを管理する場合は無効にします。

暗号化キーのローテーションを無効にする

CDK
Terraform

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

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

キーローテーションを無効にする、またはローテーションしない独自のキーを持ち込むと、Checkov の CKV_AWS_7 に失敗します。キーで抑制してください：

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table.encryptionKey!, ['CKV_AWS_7'], 'Key rotation is managed externally');

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

接続

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

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

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

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