Claude Code 与 AWS DynamoDB 实战指南：从键设计到安全写入

只对 Claude Code 说“帮我接入 DynamoDB”，很容易得到一份看起来能跑、上线后却难维护的代码。DynamoDB 不是“不用设计 schema 的数据库”。它真正的 schema 是访问模式：页面如何读取数据、API 用哪个键查询、哪些写入不能覆盖旧数据、流量是否平均分布在分区键上。

本文把 Claude Code 当作设计评审助手，而不是单纯的代码生成器。我们会从表设计、分区键、简单设计与单表设计、条件写入、TTL、本地测试、IAM、成本和热点分区讲到可复制运行的 Node.js 示例。相关内容可以继续阅读 AWS Lambda 指南、AWS IAM 指南 和 AWS CloudWatch 运用。

实现时请以 AWS 官方文档为准： DynamoDB data modeling foundations、partition key best practices、Query key condition expressions、condition expressions、TTL、DynamoDB local、throughput capacity、IAM fine-grained access control。

先让 Claude Code 画出访问模式

访问模式就是应用真实的读写动作。比如：按项目列出任务、按任务 ID 更新一个任务、让用户会话 7 天后过期、保证同一个 webhook 事件只处理一次。DynamoDB 的设计要从这些动作倒推，而不是从“需要几个字段”开始。

先把这段提示词交给 Claude Code：

请在写代码前评审这个 DynamoDB 设计。
需求:
- 按项目列出任务
- 按 taskId 更新一个任务
- 用户会话 7 天后过期
- 每个 webhook eventId 只处理一次

请输出:
1. 访问模式表
2. PK/SK 设计
3. 哪些操作可以用 Query，哪些不行
4. 需要条件写入的地方
5. 热点分区和成本风险

这样做的原因很简单：DynamoDB 的 Query 必须包含分区键的等值条件，排序键可以再用 begins_with 等条件缩小范围。FilterExpression 不是 SQL 的 WHERE 替代品，它是在读取后再过滤，容量已经消耗了。如果 Claude Code 为列表页生成 Scan，先把它当作设计问题处理。

简单设计还是单表设计

单表设计把多个实体类型放进一张表，用 PK 和 SK 的前缀区分数据。例如一个项目下面可以有 META、TASK#001、EVENT#...。它的好处是相关数据可以通过一次 Query 取到；代价是学习成本、权限边界、Streams 处理和备份策略都会更复杂。

简单设计更接近“一个用途一张表”，或者至少让一张表只承载少量相关访问模式。对 MVP、内部工具和刚开始用 Claude Code 的团队来说，简单设计更容易评审，也更容易发现错误。

判断点	简单设计	单表设计
早期开发	容易解释	需要更强评审
多实体页面	可能需要多次读取	常能一次 Query
IAM	表级隔离更直观	LeadingKeys 很重要
变更	可以新增表	键命名必须稳定
适合场景	MVP、学习、小工具	访问模式稳定的业务系统

本文示例使用一张表，但只覆盖项目、任务、会话和 webhook 去重。它不是把所有业务硬塞进单表，而是用一个小范围练习键设计。

ClaudeCodeLabDemo

PK                 SK                   entityType
PROJECT#alpha      META                 Project
PROJECT#alpha      TASK#task-001        Task
USER#u-001         SESSION#s-001        Session
WEBHOOK#stripe     EVENT#evt_001        WebhookEvent

查询方式:
- 项目任务列表: PK = PROJECT#alpha AND begins_with(SK, TASK#)
- 用户会话: PK = USER#u-001 AND begins_with(SK, SESSION#)
- Webhook 去重: 对同一个 PK/SK 做条件 PutItem

三个具体用例

第一个用例是项目任务看板。把一个项目下的任务放在 PK = PROJECT#projectId 下，SK = TASK#taskId。任务列表就是一次 Query，单个任务更新则使用完整主键。Claude Code 如果建议按 status 直接查询，就要追问是加 GSI，还是先按项目读取后在应用层处理。

第二个用例是会话、邀请链接和临时令牌。DynamoDB TTL 使用 Number 类型的 Unix epoch 秒作为过期时间。TTL 适合清理不再需要的数据，但不是精确定时任务。过期 item 可能在后台删除前仍然能被读到，所以涉及安全的数据仍要在应用层检查 expiresAt。

第三个用例是 webhook 幂等性。幂等性指同一事件重复到达也不会重复扣款、重复发邮件或重复改状态。用 WEBHOOK#provider 和 EVENT#eventId 作为键，再用 attribute_not_exists(PK) AND attribute_not_exists(SK) 做条件写入，就能让第一个写入成功，后续重复事件失败。

还可以扩展到限流。比如 PK = RATE#userId、SK = WINDOW#2026-06-03T10:00，在时间窗口内递增计数。不过热门用户或热门租户很容易形成热点分区，高流量公开 API 不应该只靠这个简单方案。

本地运行环境

先启动 DynamoDB Local。下面的 Compose 文件与 AWS 官方文档中的 Docker 方案保持一致。

services:
  dynamodb-local:
    image: "amazon/dynamodb-local:latest"
    command: "-jar DynamoDBLocal.jar -sharedDb -dbPath ./data"
    ports:
      - "8000:8000"
    volumes:
      - "./docker/dynamodb:/home/dynamodblocal/data"
    working_dir: /home/dynamodblocal

docker compose up -d
export AWS_ACCESS_KEY_ID=fakeMyKeyId
export AWS_SECRET_ACCESS_KEY=fakeSecretAccessKey
export AWS_REGION=us-west-2

创建本地表。运行前确认带有 --endpoint-url http://localhost:8000，避免误建真实 AWS 表。

aws dynamodb create-table \
  --table-name ClaudeCodeLabDemo \
  --attribute-definitions AttributeName=PK,AttributeType=S AttributeName=SK,AttributeType=S \
  --key-schema AttributeName=PK,KeyType=HASH AttributeName=SK,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --endpoint-url http://localhost:8000 \
  --region us-west-2

启用 TTL 属性。DynamoDB Local 适合验证属性形状和应用逻辑，不适合证明生产环境的删除时机。

aws dynamodb update-time-to-live \
  --table-name ClaudeCodeLabDemo \
  --time-to-live-specification "Enabled=true,AttributeName=expiresAt" \
  --endpoint-url http://localhost:8000 \
  --region us-west-2

安装依赖：

npm init -y
npm install @aws-sdk/client-dynamodb @aws-sdk/lib-dynamodb

可复制运行的实现

保存为 app.mjs。它会创建任务、查询项目任务、用条件更新完成任务，并创建一个带 TTL 的会话 item。

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
  DynamoDBDocumentClient,
  PutCommand,
  QueryCommand,
  UpdateCommand,
} from "@aws-sdk/lib-dynamodb";

const TABLE_NAME = process.env.TABLE_NAME ?? "ClaudeCodeLabDemo";
const isLocal = process.env.DDB_LOCAL !== "0";

const client = new DynamoDBClient({
  region: process.env.AWS_REGION ?? "us-west-2",
  ...(isLocal
    ? {
        endpoint: "http://localhost:8000",
        credentials: {
          accessKeyId: "fakeMyKeyId",
          secretAccessKey: "fakeSecretAccessKey",
        },
      }
    : {}),
});

const ddb = DynamoDBDocumentClient.from(client, {
  marshallOptions: { removeUndefinedValues: true },
});

const nowIso = () => new Date().toISOString();
const ttlAfterDays = (days) => Math.floor(Date.now() / 1000) + days * 86400;
const taskKey = (projectId, taskId) => ({
  PK: `PROJECT#${projectId}`,
  SK: `TASK#${taskId}`,
});

async function createTask({ projectId, taskId, title, ownerId }) {
  const item = {
    ...taskKey(projectId, taskId),
    entityType: "Task",
    title,
    ownerId,
    status: "OPEN",
    createdAt: nowIso(),
    updatedAt: nowIso(),
  };

  await ddb.send(
    new PutCommand({
      TableName: TABLE_NAME,
      Item: item,
      ConditionExpression: "attribute_not_exists(PK) AND attribute_not_exists(SK)",
    }),
  );

  return item;
}

async function listProjectTasks(projectId) {
  const result = await ddb.send(
    new QueryCommand({
      TableName: TABLE_NAME,
      KeyConditionExpression: "PK = :pk AND begins_with(SK, :taskPrefix)",
      ExpressionAttributeValues: {
        ":pk": `PROJECT#${projectId}`,
        ":taskPrefix": "TASK#",
      },
      ReturnConsumedCapacity: "TOTAL",
    }),
  );

  console.log("consumed capacity:", result.ConsumedCapacity);
  return result.Items ?? [];
}

async function completeTask({ projectId, taskId, expectedOwnerId }) {
  const result = await ddb.send(
    new UpdateCommand({
      TableName: TABLE_NAME,
      Key: taskKey(projectId, taskId),
      UpdateExpression: "SET #status = :done, updatedAt = :now",
      ConditionExpression: "ownerId = :ownerId AND #status <> :done",
      ExpressionAttributeNames: {
        "#status": "status",
      },
      ExpressionAttributeValues: {
        ":done": "DONE",
        ":ownerId": expectedOwnerId,
        ":now": nowIso(),
      },
      ReturnValues: "ALL_NEW",
    }),
  );

  return result.Attributes;
}

async function createSession({ userId, sessionId }) {
  await ddb.send(
    new PutCommand({
      TableName: TABLE_NAME,
      Item: {
        PK: `USER#${userId}`,
        SK: `SESSION#${sessionId}`,
        entityType: "Session",
        createdAt: nowIso(),
        expiresAt: ttlAfterDays(7),
      },
      ConditionExpression: "attribute_not_exists(PK) AND attribute_not_exists(SK)",
    }),
  );
}

async function main() {
  const projectId = "alpha";
  const taskId = `task-${Date.now()}`;

  await createTask({
    projectId,
    taskId,
    title: "Review DynamoDB key design",
    ownerId: "masa",
  });

  await createSession({
    userId: "masa",
    sessionId: `session-${Date.now()}`,
  });

  console.log(await listProjectTasks(projectId));
  console.log(
    await completeTask({
      projectId,
      taskId,
      expectedOwnerId: "masa",
    }),
  );
}

main().catch((error) => {
  if (error.name === "ConditionalCheckFailedException") {
    console.error("Condition failed:", error.message);
    process.exit(2);
  }

  console.error(error);
  process.exit(1);
});

DDB_LOCAL=1 node app.mjs

让 Claude Code 改成 Lambda 或 API 路由时，要明确保留关键约束：

请把 app.mjs 拆成 Lambda handler。
不要删除 ConditionExpression。
除非解释 Query 为什么不可行，否则不要添加 Scan。
expiresAt 必须保持 Unix epoch 秒的 Number。
开发环境保留 ReturnConsumedCapacity。

IAM 与安全边界

单表设计下，单纯允许整张表的 Query 和 PutItem 往往太宽。DynamoDB 的 IAM 条件键可以用 dynamodb:LeadingKeys 限制分区键。下面示例假设角色带有 projectId principal tag，实际使用时请替换账号、区域、表名和标签策略。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ProjectScopedDynamoDBAccess",
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:UpdateItem",
        "dynamodb:DeleteItem",
        "dynamodb:Query"
      ],
      "Resource": [
        "arn:aws:dynamodb:ap-northeast-1:123456789012:table/ClaudeCodeLabProd",
        "arn:aws:dynamodb:ap-northeast-1:123456789012:table/ClaudeCodeLabProd/index/*"
      ],
      "Condition": {
        "ForAllValues:StringEquals": {
          "dynamodb:LeadingKeys": [
            "PROJECT#${aws:PrincipalTag/projectId}"
          ]
        }
      }
    }
  ]
}

评审 Claude Code 生成的 IAM 时，至少确认四点：为什么需要 Scan、是否包含 index ARN、生产和开发表是否分离、ReturnValues 或投影表达式是否会暴露不该返回的属性。

成本和热点分区

新项目通常可以先用 on-demand 模式，因为它按请求计费，不需要一开始就预测读写容量。稳定流量则可以评估 provisioned 模式。不要只对 Claude Code 说“设计得便宜一点”；请给出 item 大小、读写次数、峰值形状和可接受的延迟。

常见坑包括：

• 列表页依赖 Scan
• 把 FilterExpression 当成 SQL WHERE
• 使用 GLOBAL、TODAY、DEFAULT 这种高流量固定分区键
• 以为 TTL 会在过期秒立刻删除
• 没有说明访问模式就追加 GSI
• 订单、webhook、状态流转缺少条件写入
• 把本地 endpoint 留在生产配置中

可以让 Claude Code 做一次专门审计：

请审计这个 DynamoDB 实现的 Scan 依赖、热点分区、TTL 误解、条件写入缺失、IAM 权限过大和 on-demand 成本暴涨风险。先列问题，再给修复建议。

如果你希望把这类提示词、检查清单和评审模板固定下来，可以查看 ClaudeCodeLab 产品。团队要把 AWS、IAM、CI 评审和上线验证一起整理时，可以从 Claude Code 培训与咨询 开始。

总结

DynamoDB 的关键不是“少建几张表”，而是清楚说明访问模式、键设计、失败条件和成本边界。Claude Code 能快速生成实现，但不能替你判断哪些查询会变贵、哪些键会过热、哪些写入必须具备幂等性。

实测记录：最有效的顺序是先让 Claude Code 生成访问模式表，再评审 PK/SK 和失败条件，最后在 DynamoDB Local 上运行条件写入。尤其是 webhook 去重中的 attribute_not_exists，可以直接变成生产代码评审时的安全检查点。