简介
无服务器架构让你无需管理服务器即可构建和运行应用。AWS Lambda 根据事件执行代码，API Gateway 提供完全托管的 HTTP 端点。结合 DynamoDB，你就拥有了一个可扩展、高性价比的 API 技术栈。
本教程将构建一个完整的任务管理系统 CRUD REST API。
前置要求
- AWS 账户（Free Tier 适用）
- AWS CLI 已配置（
aws configure） - SAM CLI 已安装
- Python 3.12+
- REST API 基础知识
项目初始化
# 安装 SAM CLI
pip install aws-sam-cli
# 初始化项目
sam init --runtime python3.12 --name task-api --app-template hello-world
cd task-api
项目结构
task-api/
├── template.yaml # SAM/CloudFormation 模板
├── src/
│ ├── handlers/ # Lambda 函数处理器
│ │ ├── create_task.py
│ │ ├── get_task.py
│ │ ├── list_tasks.py
│ │ ├── update_task.py
│ │ └── delete_task.py
│ └── utils/ # 公共工具模块
│ ├── dynamodb.py
│ └── response.py
├── tests/
├── requirements.txt
└── samconfig.toml
第1步：用 SAM 定义基础设施
编辑 template.yaml：
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: 无服务器任务管理 API
Globals:
Function:
Timeout: 10
Runtime: python3.12
MemorySize: 256
Environment:
Variables:
TABLE_NAME: !Ref TasksTable
Architectures:
- arm64 # Graviton2 — 更便宜更快
Resources:
# DynamoDB 表
TasksTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: tasks
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: task_id
AttributeType: S
- AttributeName: user_id
AttributeType: S
KeySchema:
- AttributeName: user_id
KeyType: HASH
- AttributeName: task_id
KeyType: RANGE
# API Gateway
TaskApi:
Type: AWS::Serverless::Api
Properties:
StageName: prod
Cors:
AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
AllowHeaders: "'Content-Type,Authorization'"
AllowOrigin: "'*'"
# Lambda 函数
CreateTaskFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: src/
Handler: handlers.create_task.handler
Events:
CreateTask:
Type: Api
Properties:
RestApiId: !Ref TaskApi
Path: /tasks
Method: POST
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref TasksTable
第2步：构建工具模块
DynamoDB 辅助模块
import os
import boto3
from boto3.dynamodb.conditions import Key
_table = None
def get_table():
"""跨 Lambda 调用复用 DynamoDB 连接。"""
global _table
if _table is None:
dynamodb = boto3.resource('dynamodb')
_table = dynamodb.Table(os.environ['TABLE_NAME'])
return _table
def put_item(item: dict) -> dict:
table = get_table()
table.put_item(Item=item)
return item
def query_by_user(user_id: str, limit: int = 50) -> list[dict]:
table = get_table()
response = table.query(
KeyConditionExpression=Key('user_id').eq(user_id),
Limit=limit,
ScanIndexForward=False
)
return response.get('Items', [])
响应辅助模块
import json
from decimal import Decimal
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return int(obj) if obj % 1 == 0 else float(obj)
return super().default(obj)
def success(body, status_code=200):
return {
'statusCode': status_code,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*',
},
'body': json.dumps(body, cls=DecimalEncoder)
}
def error(message, status_code=400):
return success({'error': message}, status_code)
第3步：实现 CRUD 处理器
创建任务
import json
import uuid
from datetime import datetime, timezone
from utils.dynamodb import put_item
from utils.response import success, error
def handler(event, context):
body = json.loads(event.get('body', '{}'))
title = body.get('title', '').strip()
if not title:
return error('title is required')
now = datetime.now(timezone.utc).isoformat()
task = {
'task_id': str(uuid.uuid4()),
'user_id': body.get('user_id', 'default'),
'title': title,
'description': body.get('description', ''),
'status': 'pending',
'priority': body.get('priority', 'medium'),
'created_at': now,
'updated_at': now,
}
put_item(task)
return success(task, 201)
第4步：本地测试
# 构建
sam build
# 启动本地 API（需要 Docker）
sam local start-api --port 3000
# 测试
curl -X POST http://localhost:3000/tasks \
-H 'Content-Type: application/json' \
-d '{"title": "学习 Lambda", "priority": "high"}'
curl http://localhost:3000/tasks
第5步：添加中间件验证
import json
import logging
import traceback
from functools import wraps
from utils.response import error
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def api_handler(func):
"""统一错误处理和日志记录的装饰器。"""
@wraps(func)
def wrapper(event, context):
logger.info(json.dumps({
'method': event.get('httpMethod'),
'path': event.get('path'),
'request_id': context.aws_request_id,
}))
try:
return func(event, context)
except json.JSONDecodeError:
return error('请求体 JSON 格式无效', 400)
except Exception as e:
logger.error(traceback.format_exc())
return error('Internal server error', 500)
return wrapper
第6步：部署到 AWS
# 首次部署（引导式）
sam deploy --guided
# 后续部署
sam deploy
第7步：添加 API Key 认证
TaskApi:
Type: AWS::Serverless::Api
Properties:
StageName: prod
Auth:
ApiKeyRequired: true
UsagePlan:
CreateUsagePlan: PER_API
Throttle:
BurstLimit: 50
RateLimit: 20
Quota:
Limit: 10000
Period: MONTH
成本估算
| 服务 | 免费额度 | 超出后费用 |
|------|----------|------------|
| Lambda | 每月100万次请求 | $0.20/百万次请求 |
| API Gateway | 每月100万次调用 | $3.50/百万次调用 |
| DynamoDB | 25GB 存储 | $0.25/GB/月 |
对于低流量 API（约1万次请求/天），月费用在免费额度内为 $0，超出后约 $3。
总结
你已经构建了一个完整的无服务器 REST API：