py#dynamodb

此生成器创建一个由 Amazon DynamoDB 支持的新 Python 项目，使用 PynamoDB 进行实体建模。它生成应用程序代码和基础设施，用于使用 AWS CDK 或 Terraform 配置和管理 DynamoDB 表，支持单表设计并通过 DynamoDB Local 内置本地开发功能。

用法

生成 DynamoDB 项目

VSCode

1. 安装 Nx Console VSCode Plugin 如果您尚未安装
2. 在VSCode中打开Nx控制台
3. 点击 Generate (UI) 在"Common Nx Commands"部分
4. 搜索 @aws/nx-plugin - py#dynamodb
5. 填写必需参数
6. 点击 Generate

CLI

pnpm

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

yarn

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

npm

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

bun

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

您还可以执行试运行以查看哪些文件会被更改

pnpm

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

yarn

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

npm

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

bun

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

• 文件夹<name>

  • __init__.py 包导出
  • client.py DynamoDB 客户端和表名解析
  • 文件夹entities

    • base.py 带有 GSI 声明的基础 PynamoDB 模型
    • example.py 示例实体定义
    • __init__.py 实体导出
• config.json 表配置，包括 GSI 定义和本地开发设置
• project.json 项目配置和构建目标

本地开发脚本在所有 DynamoDB 项目（TypeScript 和 Python）之间共享，并一次性生成到：

• 文件夹packages/common/scripts/src/dynamodb

  • create-local-table.ts 在本地 DynamoDB Local 实例中创建 DynamoDB 表
  • pull-image.ts 拉取 DynamoDB Local 镜像
  • start-container.ts 启动 DynamoDB Local 容器

基础设施

由于该生成器会根据您选择的 iacProvider 以基础设施即代码的形式输出，它将在 packages/common 目录下创建一个包含相关 CDK 构造体或 Terraform 模块的项目。

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

CDK

• 文件夹packages/common/constructs

  • 文件夹src

    • 文件夹app/ 针对特定项目/生成器的基础设施构造体

      • …
    • 文件夹core/ 被 app 目录构造体重用的通用构造体

      • …
    • index.ts 导出 app 目录构造体的入口文件
  • project.json 项目构建目标与配置

生成的项目

此项目使用 ts#project 生成器创建，因此配置了相同的构建目标。

Terraform

• 文件夹packages/common/terraform

  • 文件夹src

    • 文件夹app/ 针对特定项目/生成器的 Terraform 模块

      • …
    • 文件夹core/ 被 app 目录模块重用的通用模块

      • …
  • project.json 项目构建目标与配置

生成的项目

此项目使用 terraform#project 生成器创建，因此配置了相同的构建目标。

CDK

• 文件夹packages/common/constructs/src

  • 文件夹app

    • 文件夹dynamodb

      • <name>.ts 特定于您的表的基础设施
  • 文件夹core

    • dynamodb.ts 通用 DynamoDB 表构造

Terraform

• 文件夹packages/common/terraform/src

  • 文件夹app

    • 文件夹dynamodb

      • 文件夹<name>

        • <name>.tf 特定于您的表的模块
  • 文件夹core

    • 文件夹dynamodb

      • dynamodb.tf 通用 DynamoDB 模块

本地开发

启动本地 DynamoDB

生成器配置了一个 dev 目标，用于启动 DynamoDB Local 实例并创建表。使用项目的 dev 目标：

pnpm

pnpm nx dev <project-name>

yarn

yarn nx dev <project-name>

npm

npx nx dev <project-name>

bun

bunx nx dev <project-name>

这将自动：

1. 拉取 DynamoDB Local 镜像（pull-image 目标）
2. 启动容器
3. 使用 config.json 中定义的索引创建本地表

数据建模

生成的项目使用 PynamoDB 进行实体建模。所有实体必须继承自生成的 BaseModel — 它在运行时解析正确的 DynamoDB 表名，在部署时从 AWS AppConfig 读取，或在通过 DynamoDB Local 本地运行时从 config.json 读取。如果没有这个，PynamoDB 将不知道要使用哪个表。BaseModel 还使用 PynamoDB 的多态支持 在单个表中存储多种实体类型，遵循 DynamoDB 的单表设计。

在 <name>/entities/ 下添加或更新实体文件，使用生成的示例实体作为起点：

packages/my_table/my_table/entities/example.py

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 并非专为单表设计而构建

PynamoDB 的 DiscriminatorAttribute 是为单个同构表中的类层次结构多态性而设计的 — 而不是为在一个表中存储结构不同的实体类型。这里使用的模式是有效的，但请记住以下注意事项：

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

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

packages/my_table/my_table/entities/user.py

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)

packages/my_table/my_table/entities/order.py

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 导出新实体：

packages/my_table/my_table/entities/__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 vs 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：

packages/my_table/my_table/entities/order.py (项目集合变体)

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

# 在一个主表查询中检索用户及其所有订单
# 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，其中一篇文章可以有多个标签，一个标签可以应用于多篇文章：

packages/my_table/my_table/entities/article_tag.py

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：

• 连接 — 当 is_local() 为 True 时，BaseModel.Meta 从 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

由于 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 中的更改：

config.json

{
  ...
  "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

警告

DynamoDB 不允许在单次表更新中创建或删除多个 GSI。

每次部署最多只能添加或删除一个 GSI。如果需要添加或删除多个 GSI，请逐个进行 — 更新 config.json 和相关实体类，部署堆栈，然后对下一个更改重复此过程。

连接到表

在任何 Python 项目中，将 DynamoDB 包添加为工作区依赖项并直接导入实体类：

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

运行时配置

在 AWS 中运行时，get_table_name() 使用 RUNTIME_CONFIG_APP_ID 环境变量从 AWS AppConfig 获取表名。使用此插件构建的项目已自动配置此变量。对于其他 Python 项目，请确保在运行时环境中设置 RUNTIME_CONFIG_APP_ID，其值为基础设施配置的 AppConfig 应用程序 ID。有关更多信息，请参阅运行时配置指南。

部署您的表

DynamoDB 生成器会根据您选择的 iac 创建 CDK 或 Terraform 基础设施。

CDK

CDK 构造在 common/constructs 中创建。使用示例：

packages/infra/src/stacks/application-stack.ts

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

Terraform 模块在 common/terraform 中创建。使用示例：

packages/infra/src/main.tf

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

这将配置一个具有以下特性的 DynamoDB 表：

• pk（分区键）和 sk（排序键），均为 String 类型
• 在 config.json 中定义的全局二级索引
• 按需（PAY_PER_REQUEST）计费
• 客户管理的 KMS 加密，并自动轮换密钥
• 启用时间点恢复
• 启用删除保护
• 表名注册到运行时配置中

授予访问权限

要允许 Lambda 函数访问 DynamoDB 表，请在您的基础设施中授予必要的权限。

CDK

在表构造上调用 grantReadWriteData。这将授予 Lambda 执行角色所需的 DynamoDB 和 KMS 权限：

packages/infra/src/stacks/application-stack.ts

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);
});

注意

此示例授予 API 中的每个处理程序访问权限。如果只有某些处理程序需要访问权限，请单独配置它们。

Terraform

授予 Lambda 执行角色访问 DynamoDB 表及其 KMS 加密密钥的权限：

packages/infra/src/main.tf

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

packages/infra/src/stacks/application-stack.ts

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

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

Terraform

packages/infra/src/main.tf

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

计费模式

表默认使用按需（PAY_PER_REQUEST）计费。对于可预测的高吞吐量工作负载，可以切换到预配置容量。

CDK

packages/infra/src/stacks/application-stack.ts

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,
});

Terraform

packages/infra/src/main.tf

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

时间点恢复

默认启用时间点恢复，允许您将表恢复到过去 35 天内的任何时间点。

禁用时间点恢复

CDK

packages/infra/src/stacks/application-stack.ts

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

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

Terraform

packages/infra/src/main.tf

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

加密密钥轮换

用于加密表的 KMS 密钥默认启用自动密钥轮换。如果您的安全策略在外部管理轮换，可以禁用它。

禁用加密密钥轮换

CDK

packages/infra/src/stacks/application-stack.ts

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

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

Terraform

packages/infra/src/main.tf

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 表