Python DynamoDB

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

使用方法

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

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

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

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

オプション

パラメータ	型	デフォルト	説明
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 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 テーブルコンストラクト

ローカル開発

ローカル 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名前空間下のランタイム設定にテーブル名が登録される

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

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

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

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

削除保護

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

削除保護の無効化

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

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 DynamoDBFastAPI を DynamoDB テーブルに接続する

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

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