基于 Serverless Framework 的 Python HTTP API + DynamoDB 实战：Todo 增删改查全流程解析

本文围绕开源仓库 examples 中的 aws-python-http-api-with-dynamodb 示例，系统讲解如何利用 Serverless Framework 在 AWS 上构建一个支持创建、查询、更新、删除（CRUD）Todo 的 HTTP API，并以 DynamoDB 作为数据存储后端。读完本文，你将掌握 serverless.yml 的完整部署配置、五个 Lambda 函数的实现思路与底层调用链、通过 curl 实测 API 的完整姿势，以及 Lambda 并发与 DynamoDB 容量规划的关键要点。

一、示例概览：HTTP API + DynamoDB 的 Todo 服务

该示例的核心目标是：搭建一个 HTTP API，允许对 Todo 进行 create（创建）、list（列出）、get（单个查询）、update（更新）、delete（删除）五种操作，数据统一存放在 DynamoDB 表中。示例作者在文档开头也特别说明：DynamoDB 只是存储后端的一个示例选择，实际项目中完全可以替换为任意数据存储。

示例明确列出的两类典型应用场景：

• Web 应用的 API 后端；
• 移动应用的 API 后端。

从架构上看，这个服务属于典型的"API Gateway（HTTP API）+ Lambda + DynamoDB"无服务器三层结构：客户端通过 HTTP 请求触发 API 事件，Serverless Framework 将请求路由到对应的 Lambda 函数，函数通过 boto3 访问 DynamoDB 完成数据读写。

二、目录结构与"一操作一文件"设计思想

整个服务的函数代码集中在 todos 目录下，其设计原则非常清晰：

• 每个操作对应一个文件，例如 todos/delete.py；
• 每个文件内只定义一个函数，例如 create 函数、list 函数等。

目录布局如下：

aws-python-http-api-with-dynamodb/
├── todos/
│   ├── __init__.py
│   ├── create.py        # 创建 Todo
│   ├── list.py          # 列出全部 Todo
│   ├── get.py           # 按 id 获取单个 Todo
│   ├── update.py        # 更新 Todo
│   ├── delete.py        # 删除 Todo
│   └── decimalencoder.py # Decimal 序列化工具
├── serverless.yml       # 部署配置
├── package.json
└── README.md

原文档特别解释了这一目录结构背后的扩展思路：todos 目录的用意是，如果你想在同一个服务里承载多个资源（例如 users、notes、comments），可以把它们放在同一个服务的不同目录下。虽然技术上完全可行，但作者也建议：每个资源单独拆分为一个服务可能更清晰，具体取决于实际用例和个人偏好。这是无服务器架构中"单体服务 vs 细粒度服务"的经典取舍，也是本示例在目录设计上想要传递的工程思想。

三、部署配置深度解析：serverless.yml 逐段拆解

部署配置位于 serverless.yml，它完整定义了服务名、运行时、函数事件映射、IAM 权限与 DynamoDB 表资源。下面逐段解读。

3.1 服务声明与框架版本

service: serverless-http-api-dynamodb
frameworkVersion: '4'

• service：服务名，同时被用作函数名与 DynamoDB 表名的前缀，形成命名空间；
• frameworkVersion: '4'：声明本项目基于 Serverless Framework v4 编写（当前仓库的示例均以此为目标框架版本）。

3.2 provider 配置：运行时、架构与环境变量

provider:
  name: aws
  runtime: python3.14
  architecture: arm64
  environment:
    DYNAMODB_TABLE: ${self:service}-${sls:stage}
  httpApi:
    cors: true
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:Query
            - dynamodb:Scan
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
            - dynamodb:DeleteItem
          Resource: "arn:aws:dynamodb:${aws:region}:*:table/${self:provider.environment.DYNAMODB_TABLE}"

关键点说明：

• runtime: python3.14：Lambda 函数运行环境为 Python 3.14；
• architecture: arm64：选择 ARM 架构（Graviton），通常比 x86 更经济；
• DYNAMODB_TABLE：通过变量引用 self:service（服务名）与 sls:stage（部署阶段）动态拼出表名，例如默认 dev 阶段下表名为 serverless-http-api-dynamodb-dev。这样同一份配置在不同 stage（dev/prod）下会自动使用不同的表，避免数据串扰；
• httpApi.cors: true：一键为 HTTP API 开启 CORS 支持，方便浏览器端（Web/移动前端）直接跨域调用；
• iam.role.statements：为函数授予的最小权限集——Query、Scan、GetItem、PutItem、UpdateItem、DeleteItem 六项 DynamoDB 操作，且 Resource 通过 ARN 精确限定到当前服务对应的那张表，遵循最小权限原则。

3.3 functions：五个函数与 HTTP 事件映射

functions:
  create:
    handler: todos/create.create
    events:
      - httpApi:
          path: /todos
          method: post
  list:
    handler: todos/list.list
    events:
      - httpApi:
          path: /todos
          method: get
  get:
    handler: todos/get.get
    events:
      - httpApi:
          path: /todos/{id}
          method: get
  update:
    handler: todos/update.update
    events:
      - httpApi:
          path: /todos/{id}
          method: put
  delete:
    handler: todos/delete.delete
    events:
      - httpApi:
          path: /todos/{id}
          method: delete

事件映射遵循 REST 惯例：

函数	Handler	路径	方法	用途
create	todos/create.py	/todos	POST	创建
list	todos/list.py	/todos	GET	列出全部
get	todos/get.py	/todos/{id}	GET	单个查询
update	todos/update.py	/todos/{id}	PUT	更新
delete	todos/delete.py	/todos/{id}	DELETE	删除

注意 handler 的写法：todos/create.create 表示模块路径 todos/create.py 中的 create 函数——这正是"一个文件一个函数"约定在部署层落地的方式。路径参数 {id} 会出现在函数收到的 event['pathParameters']['id'] 中（见第五、七节）。

3.4 resources：DynamoDB 表定义

resources:
  Resources:
    TodosDynamoDbTable:
      Type: 'AWS::DynamoDB::Table'
      DeletionPolicy: Retain
      Properties:
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        BillingMode: PAY_PER_REQUEST
        TableName: ${self:provider.environment.DYNAMODB_TABLE}

• 主键设计：以 id（字符串 S 类型）作为唯一的哈希键（HASH），所有 get/update/delete 都按该主键精确定位；
• BillingMode: PAY_PER_REQUEST：采用按需计费模式，表容量随流量自动伸缩，无需预设读写容量，适合流量波动大的场景；
• DeletionPolicy: Retain：删除 CloudFormation 堆栈时保留 DynamoDB 表及其数据，避免误删生产数据。

package.json（查看）只是最简单的元信息声明（MIT 许可），本示例不依赖任何 npm 包，纯 Python + boto3 即可运行。

四、环境准备与部署上线

4.1 安装 Serverless Framework

在本地安装全局 CLI（Node.js 环境需就绪）：

npm install -g serverless

4.2 执行部署

在项目目录（即 aws-python-http-api-with-dynamodb）下运行：

serverless deploy

原文档给出了预期的部署输出，关键信息如下：

Service Information
service: serverless-http-api-dynamodb
stage: dev
region: us-east-1
endpoints:
  POST - https://45wf34z5yf.execute-api.us-east-1.amazonaws.com/todos
  GET - https://45wf34z5yf.execute-api.us-east-1.amazonaws.com/todos
  GET - https://45wf34z5yf.execute-api.us-east-1.amazonaws.com/todos/{id}
  PUT - https://45wf34z5yf.execute-api.us-east-1.amazonaws.com/todos/{id}
  DELETE - https://45wf34z5yf.execute-api.us-east-1.amazonaws.com/todos/{id}
functions:
  update: serverless-http-api-dynamodb-dev-update
  get: serverless-http-api-dynamodb-dev-get
  list: serverless-http-api-dynamodb-dev-list
  create: serverless-http-api-dynamodb-dev-create
  delete: serverless-http-api-dynamodb-dev-delete

从输出可以直观确认：服务自动进入 dev 阶段、部署到 us-east-1 区域（可在配置或部署参数中调整），API Gateway 生成了 5 个 REST 风格端点，5 个 Lambda 函数命名统一为 服务名-阶段-函数名 格式。部署完成后，即可用下文 curl 命令实测接口。

五、五个 Lambda 函数的源码实现剖析

所有函数都遵循统一模式：boto3.resource('dynamodb') 初始化客户端，从 os.environ['DYNAMODB_TABLE'] 读取表名，完成数据操作后返回带 statusCode 与 body 的标准 Lambda 响应结构。

5.1 create：写入新 Todo

实现见 todos/create.py，核心流程：

1. json.loads(event['body']) 解析请求体；
2. 参数校验：若请求体中不含 text 字段，记录错误日志并抛异常（"Couldn't create the todo item."）；
3. 以 uuid.uuid1() 生成主键 id；
4. 构造完整 item：text（内容）、checked: False（默认未完成）、createdAt 与 updatedAt 均为 str(time.time())（字符串形式的秒级时间戳）；
5. table.put_item(Item=item) 写入 DynamoDB；
6. 返回 statusCode: 200 与 json.dumps(item)。

值得注意：虽然原文档在"Create a Todo"一节标注为 "No output"，但从源码看，create 函数实际会以 200 状态码返回新创建 Todo 的完整 JSON 内容——这是 README 描述与真实实现的一处差异，实测调用时请以实际返回为准。

5.2 list：Scan 全表

实现见 todos/list.py：直接执行 table.scan() 拉取全部条目，并以 json.dumps(result['Items'], cls=decimalencoder.DecimalEncoder) 序列化返回。

这里引入了本示例的关键工具类 DecimalEncoder（见 todos/decimalencoder.py）：DynamoDB 返回的数字类型在 boto3 中表现为 decimal.Decimal，而标准 JSON 序列化无法直接处理它（这是 Python bug 16535 的经典场景），因此自定义编码器将 Decimal 转为 int 后再输出。这也解释了为什么 list 与 get 都要显式传入 cls=decimalencoder.DecimalEncoder。

5.3 get：按主键查询单个 Todo

实现见 todos/get.py：从 event['pathParameters']['id'] 取路径参数，执行 table.get_item(Key={'id': ...})，返回 result['Item'] 的 JSON 序列化结果。由于 id 正是表的主键（HASH 键），该查询按主键精确定位，效率最高。

5.4 update：UpdateExpression 局部更新

实现见 todos/update.py，流程比 create 更丰富：

1. 解析请求体并校验 text、checked 两个字段必须同时存在；
2. 时间戳改用 int(time.time() * 1000)（毫秒级整数）——与 create 的秒级字符串时间戳格式不同，这是本示例实现中两个函数时间戳格式不一致的细节，实际使用需自行统一；
3. 通过 update_item 执行表达式更新：

table.update_item(
    Key={'id': event['pathParameters']['id']},
    ExpressionAttributeNames={'#todo_text': 'text'},
    ExpressionAttributeValues={
        ':text': data['text'],
        ':checked': data['checked'],
        ':updatedAt': timestamp,
    },
    UpdateExpression='SET #todo_text = :text, '
                     'checked = :checked, '
                     'updatedAt = :updatedAt',
    ReturnValues='ALL_NEW',
)

• ExpressionAttributeNames 中的 #todo_text 是 text 的占位别名（因为 text 可能触及保留字/特殊语义，使用别名更安全）；
• UpdateExpression 用表达式而非整条覆盖，实现真正的局部更新；
• ReturnValues='ALL_NEW' 让更新操作直接返回更新后的完整条目，响应体即最新数据。

5.5 delete：按主键删除

实现见 todos/delete.py：执行 table.delete_item(Key={'id': ...})，返回仅含 statusCode: 200 的响应体，无内容输出——与原文档 "No output" 的描述一致。

5.6 小结：五个函数的分工对照

函数	DynamoDB 操作	入参来源	响应内容
create	PutItem	event['body']	创建的 item
list	Scan	—	全部 items
get	GetItem	event['pathParameters']['id']	单个 item
update	UpdateItem	body + 路径 id	更新后的 item（ALL_NEW）
delete	DeleteItem	event['pathParameters']['id']	仅状态码

六、API 使用指南：curl 实战

部署完成后，用下列命令对端点（https://XXXXXXX.execute-api.us-east-1.amazonaws.com 为你的实际部署地址）进行增删改查。

6.1 创建 Todo

curl -X POST https://XXXXXXX.execute-api.us-east-1.amazonaws.com/todos --data '{ "text": "Learn Serverless" }' -H "Content-Type: application/json"

如上文所述，实际会返回新创建的 Todo（含 id、checked: false、createdAt/updatedAt）。

6.2 列出全部 Todos

curl https://XXXXXXX.execute-api.us-east-1.amazonaws.com/todos

原文档给出的示例输出（注意末尾的 % 是终端回显，非响应内容）：

[{"text":"Deploy my first service","id":"ac90feaa11e6-9ede-afdfa051af86","checked":true,"updatedAt":1479139961304},{"text":"Learn Serverless","id":"206793aa11e6-9ede-afdfa051af86","createdAt":1479139943241,"checked":false,"updatedAt":1479139943241}]

可以看到 list 返回的是 JSON 数组，包含多笔记录及各自的 text、id、checked、时间戳字段。

6.3 获取单个 Todo

# 将 <id> 替换为 todos 表中真实的 id
curl https://XXXXXXX.execute-api.us-east-1.amazonaws.com/todos/<id>

示例结果：

{"text":"Learn Serverless","id":"ee6490d0-aa11e6-9ede-afdfa051af86","createdAt":1479138570824,"checked":false,"updatedAt":1479138570824}

6.4 更新 Todo

# 将 <id> 替换为 todos 表中真实的 id
curl -X PUT https://XXXXXXX.execute-api.us-east-1.amazonaws.com/todos/<id> --data '{ "text": "Learn Serverless", "checked": true }' -H "Content-Type: application/json"

请求体必须同时携带 text 与 checked（源码校验），示例结果中 checked 已变为 true，updatedAt 同步刷新：

{"text":"Learn Serverless","id":"ee6490d0-aa11e6-9ede-afdfa051af86","createdAt":1479138570824,"checked":true,"updatedAt":1479138570824}

6.5 删除 Todo

# 将 <id> 替换为 todos 表中真实的 id
curl -X DELETE https://XXXXXXX.execute-api.us-east-1.amazonaws.com/todos/<id>

无输出内容，仅返回 200 状态码。

七、事件对象关键字段速查

无论使用哪个 API 客户端，Lambda 收到的 event 对象中与本示例直接相关的字段有：

字段	来源	用途
event['body']	请求体	create/update 解析 JSON 负载
event['pathParameters']['id']	URL 路径 {id} 段	get/update/delete 的主键定位
os.environ['DYNAMODB_TABLE']	部署时注入的环境变量	定位目标表（如 serverless-http-api-dynamodb-dev）

其中 DYNAMODB_TABLE 由 serverless.yml 中的 ${self:service}-${sls:stage} 变量在部署时自动解析，函数代码只需读取环境变量即可，无需硬编码表名，从而保证多环境复用。

八、容量规划与伸缩要点

8.1 AWS Lambda 并发限制

原文档指出：默认情况下，AWS Lambda 将某个区域内所有函数的总并发执行数限制为 1000。该默认限制是保护性安全阈值，用于防止开发和测试初期因失控或递归函数产生不可控成本。如需提升上限，需要按照 AWS 官方流程申请提升并发限制（参见 Lambda 文档中 "request a limit increase for concurrent executions" 相关章节）。

8.2 DynamoDB 容量模式：按需计费 vs 预置吞吐

原文档在缩放章节给出的预置吞吐示例为：

ProvisionedThroughput:
  ReadCapacityUnits: 1
  WriteCapacityUnits: 1

即在 serverless.yml 的 TodosDynamoDbTable.Properties 下配置预置读写容量单位（RCU/WCU），DynamoDB 会预留相应资源以保障一致、低延迟的性能，且支持随时增减。

需要特别说明：当前仓库版本的 serverless.yml 实际采用的是 BillingMode: PAY_PER_REQUEST（按需计费），即无需预设容量、随实际读写请求自动伸缩。两种模式各有适用场景：

• PAY_PER_REQUEST（按需）：适合流量不可预测、波动明显的应用，按实际请求量计费，无需容量规划；
• ProvisionedThroughput（预置）：适合流量可预测、需要精细成本控制的应用，可显式设置读写容量并随时调整。

若预期流量存在大幅波动，原文档还建议关注 DynamoDB 的自动扩缩容方案（即 AWS 官方博客介绍的 Dynamic DynamoDB 一类的做法），以在预置模式下按需动态调整容量。

九、小结

本示例用最精简的结构串联起了无服务器开发的核心链路：

1. 配置层：一份 serverless.yml 同时完成 API 路由、IAM 最小权限与 DynamoDB 表资源声明；
2. 实现层：todos 目录下五个独立文件对应五个函数，职责单一、易于扩展（可在同一服务中加入 users/notes/comments，或按资源拆分独立服务）；
3. 调用层：标准的 REST 端点 + curl 即可完成全部增删改查，返回体统一为 JSON；
4. 运维层：默认按需计费的表模式与最小权限 IAM 体现了成本与安全的最佳实践，Lambda 并发与表容量是上线前需要关注的伸缩边界。

无论是作为 Web 应用还是移动应用的 API 后端，这套"HTTP API + Python Lambda + DynamoDB"的样板都提供了可以直接复制改造的完整起点；后续在此基础上叠加鉴权、日志、前端托管等能力，即可演进为生产级无服务器应用。