用 AWS SAM 做一个网站留言接口：POST /feedback 接收 JSON，校验后写入 DynamoDB

用 AWS SAM 做一个网站留言接口：POST /feedback 接收 JSON，校验后写入 DynamoDB。HTTP API 负责跨域和限流，Lambda 只有写这一张表的权限，日志保留 14 天。

先看清成品

本机执行结果：app.py 用 moto 模拟 DynamoDB 跑了 4 个测试全部通过；template.yaml 通过 cfn-lint 1.58.0、sam validate --lint（SAM CLI 1.169.0）和 sam build。没有 sam deploy 到真实账号，也没有运行 sam local invoke（需要 Docker 和可访问的 DynamoDB）。

注意：AWS 按用量计费，价格以 AWS 官方价格页和账单为准，本文不列具体金额；做完实验记得删除资源。不要共享 root 账号或访问密钥，不要把密钥提交到仓库。中国大陆的 AWS 中国区（amazonaws.cn）是独立运营的分区，账号不通用，部分服务和 ARN 前缀（arn:aws-cn）不同，本文按全球区域编写。

隐私提示：留言里可能有邮箱等个人信息。上线前写清隐私说明，只收必要字段，并定期清理旧数据。

做完之后你会得到什么？

成果	说明
HTTP API	`https://xxx.execute-api.<区域>.amazonaws.com/feedback`
Lambda	Python 3.13，arm64，128 MB，5 秒超时
DynamoDB 表	按需计费
安全设置	CORS 限定来源、限流、最小权限、蜜罐字段

需要准备哪些环境？

项目	要求	说明
AWS 账号	有 CloudFormation、Lambda、DynamoDB、API Gateway 权限	不用 root
SAM CLI	本文用 1.169.0	按官方安装文档
Python	3.13（与运行时一致）	`sam build` 用本机 Python 构建
测试依赖	`pip install pytest boto3 moto`	本地测试

文件结构

20-aws-B-src/
├── template.yaml
├── src/
│   ├── app.py
│   └── requirements.txt
├── events/post.json
└── tests/test_app.py

第 1 步：SAM 模板

template.yaml：

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: 留言反馈接口（HTTP API + Lambda + DynamoDB）

Parameters:
  AllowedOrigin:
    Type: String
    Default: "https://example.com"
    Description: 允许跨域调用的网站地址

Globals:
  Function:
    Runtime: python3.13
    Timeout: 5
    MemorySize: 128
    Architectures: [arm64]

Resources:
  FeedbackTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH

  FeedbackApi:
    Type: AWS::Serverless::HttpApi
    Properties:
      CorsConfiguration:
        AllowOrigins: [!Ref AllowedOrigin]
        AllowMethods: [POST]
        AllowHeaders: [content-type]
      DefaultRouteSettings:
        ThrottlingBurstLimit: 10
        ThrottlingRateLimit: 5

  FeedbackFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.handler
      Environment:
        Variables:
          TABLE_NAME: !Ref FeedbackTable
      Policies:
        - DynamoDBWritePolicy:
            TableName: !Ref FeedbackTable
      Events:
        Post:
          Type: HttpApi
          Properties:
            ApiId: !Ref FeedbackApi
            Path: /feedback
            Method: POST

  FeedbackLogs:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/lambda/${FeedbackFunction}
      RetentionInDays: 14

Outputs:
  FeedbackUrl:
    Value: !Sub https://${FeedbackApi}.execute-api.${AWS::Region}.amazonaws.com/feedback

要点：

DynamoDBWritePolicy 是 SAM 内置策略模板，只授予这张表的写操作。
ThrottlingBurstLimit / ThrottlingRateLimit 限制突发和每秒请求数，防止被刷。
显式创建日志组并设保留期，否则 Lambda 日志默认永久保存。

第 2 步：函数代码

src/app.py：

import json, os, re, time, uuid
import boto3

_table = None
EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

def table():
    global _table
    if _table is None:
        _table = boto3.resource("dynamodb").Table(os.environ["TABLE_NAME"])
    return _table

def reply(code, body):
    return {"statusCode": code, "headers": {"content-type": "application/json; charset=utf-8"},
            "body": json.dumps(body, ensure_ascii=False)}

def validate(data):
    errors = []
    msg = str(data.get("message", "")).strip()
    if not 1 <= len(msg) <= 1000: errors.append("message 长度应为 1-1000")
    email = str(data.get("email", "")).strip()
    if email and not EMAIL.match(email): errors.append("email 格式不对")
    if data.get("website"): errors.append("spam")  # 蜜罐字段，正常用户看不到
    return msg, email, errors

def handler(event, context):
    try:
        data = json.loads(event.get("body") or "{}")
    except json.JSONDecodeError:
        return reply(400, {"error": "请求体不是 JSON"})
    msg, email, errors = validate(data)
    if errors:
        return reply(400, {"error": "；".join(errors)})
    item = {"id": str(uuid.uuid4()), "message": msg, "created_at": int(time.time())}
    if email: item["email"] = email
    table().put_item(Item=item)
    return reply(201, {"id": item["id"]})

website 是蜜罐字段：前端用 CSS 隐藏，正常用户不会填，机器人常会填。

第 3 步：本地测试

tests/test_app.py：

import json, os, sys
import boto3, pytest
from moto import mock_aws
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
import app

@pytest.fixture
def ddb(monkeypatch):
    monkeypatch.setenv("AWS_DEFAULT_REGION", "ap-northeast-1")
    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing"); monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
    monkeypatch.setenv("TABLE_NAME", "fb")
    with mock_aws():
        t = boto3.resource("dynamodb").create_table(TableName="fb", BillingMode="PAY_PER_REQUEST",
            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}])
        app._table = None
        yield t

def call(body): return app.handler({"body": body}, None)

def test_ok(ddb):
    ev = json.load(open(os.path.join(os.path.dirname(__file__), "..", "events", "post.json")))
    r = app.handler(ev, None); assert r["statusCode"] == 201
    item = ddb.get_item(Key={"id": json.loads(r["body"])["id"]})["Item"]
    assert item["message"] == "希望周末延长营业"

def test_bad_json(ddb): assert call("{oops")["statusCode"] == 400
def test_empty_and_email(ddb):
    r = call(json.dumps({"message": "", "email": "x"})); assert r["statusCode"] == 400
    assert "message" in r["body"] and "email" in r["body"]
def test_honeypot(ddb): assert call(json.dumps({"message": "hi", "website": "x"}))["statusCode"] == 400

pip install pytest boto3 moto
python -m pytest -q tests

检查结果：4 passed。

events/post.json 是 HTTP API 2.0 格式的样例事件：

{"version": "2.0", "routeKey": "POST /feedback", "rawPath": "/feedback", "headers": {"content-type": "application/json"}, "body": "{\"message\": \"希望周末延长营业\", \"email\": \"a@example.com\"}", "isBase64Encoded": false}

第 4 步：校验与构建

sam validate --lint
sam build

检查结果（本机实际输出节选；sam build 成功后会列出后续可用命令）：

template.yaml is a valid SAM Template
Commands you can use next
=========================
[*] Validate SAM template: sam validate
[*] Invoke Function: sam local invoke

第 5 步：部署（本文未执行）

sam deploy --guided --parameter-overrides AllowedOrigin=https://你的网站

向导会问栈名、区域、是否允许创建 IAM 角色等，答完会写入 samconfig.toml，以后直接 sam deploy。

部署后测试：

URL=$(aws cloudformation describe-stacks --stack-name 你的栈名 \
  --query "Stacks[0].Outputs[?OutputKey=='FeedbackUrl'].OutputValue" --output text)
curl -s -X POST "$URL" -H 'content-type: application/json' -d '{"message":"测试留言"}'

预期结果：返回 {"id": "..."}，状态码 201；DynamoDB 控制台能看到这条记录。

第 6 步：前端调用

<form id="fb">
  <textarea name="message" maxlength="1000" required></textarea>
  <input name="email" type="email" placeholder="邮箱（选填）">
  <input name="website" style="display:none" tabindex="-1" autocomplete="off">
  <button>提交</button>
</form>
<script>
document.getElementById("fb").addEventListener("submit", async e => {
  e.preventDefault();
  const r = await fetch("你的 FeedbackUrl", {
    method: "POST",
    headers: {"content-type": "application/json"},
    body: JSON.stringify(Object.fromEntries(new FormData(e.target)))
  });
  alert(r.ok ? "已收到，谢谢" : (await r.json()).error);
});
</script>

调试示例

提交 {"message": "", "email": "x"}，接口返回 400：

{"error": "message 长度应为 1-1000；email 格式不对"}

测试 test_empty_and_email 验证了两个错误会同时返回，用户一次就能改完。

出错了怎么办？

现象	原因	解决
浏览器报 CORS 错误	`AllowedOrigin` 与网站地址不一致（含协议和端口）	部署时传正确的地址
500 + 日志里 AccessDeniedException	函数权限不足	检查 Policies 里的表名
`sam build` 报找不到 python3.13	本机 Python 版本不同	安装 3.13，或用 `sam build --use-container`
部署时报日志组已存在	函数先被调用过，Lambda 自动建了日志组	删掉旧日志组再部署
429 Too Many Requests	触发限流	正常现象；按需调整限流参数

清理

sam delete --stack-name 你的栈名

模板规则

权限用 SAM 策略模板，不写 *。
每个函数都设日志保留期。
输入校验写在函数里并有测试，不依赖前端。

项目资源

用途	地址
AWS SAM 开发者指南	https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/what-is-sam.html
安装 SAM CLI	https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/install-sam-cli.html
Lambda 运行时	https://docs.aws.amazon.com/lambda/latest/dg/lambda-runtimes.html
API Gateway HTTP API	https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api.html
moto	https://docs.getmoto.org/en/latest/
cfn-lint	https://github.com/aws-cloudformation/cfn-lint
AWS 价格	https://aws.amazon.com/pricing/

这个接口很小，但无服务器应用的几个关键点都在：模板即基础设施、最小权限、限流、日志保留期，以及不依赖云端就能跑的测试。