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

文件夹<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 表构造

本地开发

启动本地 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 加密
启用时间点恢复
启用删除保护
表名注册到 Runtime Config

删除保护

默认启用删除保护以防止意外删除表。

禁用删除保护

对于预期会删除表的环境（例如短期开发或预览堆栈），可以禁用它。

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 表