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> 目录中创建以下项目结构：

文件夹<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
- 文件夹entities
  - 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）之间共享，并生成一次到：

文件夹packages/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 提供基础设施即代码，它将在 packages/common 中创建一个项目，其中包含相关的 CDK 构造或 Terraform 模块。

通用基础设施即代码项目的结构如下：

CDK
Terraform

文件夹packages/common/constructs
- 文件夹src
  - 文件夹app/ Constructs for infrastructure specific to a project/generator
    …
  - 文件夹core/ Generic constructs which are reused by constructs in app
    …
  - index.ts Entry point exporting constructs from app
- project.json Project build targets and configuration

文件夹packages/common/terraform
- 文件夹src
  - 文件夹app/ Terraform modules for infrastructure specific to a project/generator
    …
  - 文件夹core/ Generic modules which are reused by modules in app
    …
- project.json Project build targets and configuration

CDK
Terraform

文件夹packages/common/constructs/src
- 文件夹app
  - 文件夹dynamodb
    <name>.ts 特定于您的表的基础设施
- 文件夹core
  - dynamodb.ts 通用 DynamoDB 表构造

架构

已部署的项目配置表本身，任何连接到它的项目都可以读取和写入：

Loading the diagram…

本地开发

启动本地 DynamoDB

生成器配置了一个 dev 目标，用于启动 DynamoDB Local 实例并创建表。使用项目的 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 演示了三种模式：

按 ID 获取 —— 主索引，pk=EXAMPLE#<id>，sk=EXAMPLE#<id>
按类别列出 —— gsi1，pk=CATEGORY#<category>
按创建日期列出 —— gsi2，pk=EXAMPLE，排序键在 ISO 时间戳之间

类型前缀约定（例如 EXAMPLE#、CATEGORY#）是有意为之的：它使浏览表时项目具有自描述性，防止共享索引的实体类型之间意外的键冲突，并允许使用 begins_with 进行排序键前缀过滤。

在编写新实体之前，在文档字符串中预先定义其键模式。下一节中的 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 是为单个同构表中的类层次结构多态性设计的 —— 而不是在一个表中存储结构不同的实体类型。这里使用的模式有效，但请记住以下注意事项：

键冲突是您的责任。 PynamoDB 不会阻止两种实体类型写入相同的 pk/sk 对。始终使用唯一的类型前缀（例如 ORDER#、USER#）。
跨实体查询需要 BaseModel。 PynamoDB 在查询特定子类时会自动添加鉴别器过滤器，因此同一键上其他实体类型的项目永远不会被返回。要从共享的 GSI 分区键检索混合类型，始终通过 BaseModel 查询。
共享表设置。 所有实体共享在 BaseModel 中定义的 Meta 配置（表名、区域、凭证）。

下面是一个完整的双实体示例 —— 一个 UserModel 及其关联的 OrderModel 记录存储在同一个表中：

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 提供两个共享的 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，您可以为其分配 gsi1pk=USER#<user_id> 和 REVIEW#<id> 排序键 —— 不需要额外的 GSI。然后通过 BaseModel 查询 gsi1 会在一个请求中返回该用户的订单和评论，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 中仅检索一种实体类型，请使用排序键前缀条件：

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 导出两个关键实用程序：

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 从 config.json 设置 host，并在 is_local() 为 True 时硬编码 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 中添加或删除相应的 GlobalSecondaryIndex 类和属性来反映 BaseModel 中的更改：

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

对于仅使用哈希键的 GSI，sortKey 字段是可选的。

此配置文件是所有使用者读取的单一事实来源：

本地开发 — dev 读取 config.json 并创建或更新本地表以匹配 GSI 列表
CDK — 构造在合成时读取 config.json，因此 GSI 更改会在下次 cdk deploy 时反映出来
Terraform — 模块在计划/应用时读取 config.json

每次部署一个 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 —— 否则不会为其创建配置文件，生成的表客户端将无法解析表名。

删除保护

表由两个独立的保护机制保护，因此单独关闭其中任何一个都无法删除您的数据：

CDK
Terraform

deletionProtection，由 DynamoDB 强制执行。
RemovalPolicy.RETAIN，由 CloudFormation 强制执行，当表从堆栈中移除时会将其保留在原处。

deletion_protection_enabled，由 DynamoDB 强制执行。
lifecycle { prevent_destroy = true } 在 common/terraform/src/core/dynamodb/dynamodb.tf 中的表上，由 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 切换

在已部署的表上，在单次 terraform apply 中将 encryption 从 CUSTOMER_MANAGED 更改为 AWS_MANAGED 或 DEFAULT 会失败：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 DynamoDB将 FastAPI 连接到 DynamoDB 表

Python Agent to Python DynamoDB将 Python Agent 连接到 DynamoDB 表

Python MCP Server to Python DynamoDB将 Python MCP Server 连接到 DynamoDB 表