企业官网建设流程全解析

• 示例工程
• 教程
• 后端

导读

本文以 AWS 官方代码示例仓库中的 Aurora Item Tracker 项目为主线，讲解如何用 AWS SDK for Python（Boto3）搭建一个基于 Flask 的 REST 服务：通过 Amazon RDS Data Service 对 Aurora Serverless 数据库中的工作项进行增、查、改（归档）操作，用 AWS Secrets Manager 保管数据库凭据，并用 Amazon SES 发送工作项邮件报告。读完本文，你将掌握 Aurora Serverless 环境下的完整服务搭建流程、REST API 路由设计、参数校验与字段转换、以及基于数据量自适应的邮件发送策略。示例代码位于 python/cross_service/aurora_item_tracker，配套的 React 客户端为 resources/clients/react/elwing。

项目概览：一个跨服务的全栈示例

该项目演示了如何在一条完整链路中组合多个 AWS 服务：

• Flask REST 服务：托管本地 Web 服务，对外提供GET /api/items、POST /api/items、PUT /api/items/<id>:archive、POST /api/items:report等端点，用于列出、新增、归档工作项以及发送邮件报告。
• Amazon Aurora Serverless 数据库：存储工作项数据，通过Amazon RDS Data Service（rds-data）以 SQL 语句方式读写，无需常驻数据库连接。
• AWS Secrets Manager：保存数据库管理员凭据，RDS Data Service 通过 secret ARN 自动完成数据库身份认证。
• Amazon SES：把活动工作项以邮件形式发送给指定收件人，收件人邮箱需先在 SES 中完成验证。

该 REST 服务与 Elwing React 客户端 配合使用，在 Elwing 中通过"Item Tracker"插件即可呈现完整可操作的 Web 应用界面。

⚠️注意事项

• 运行此代码可能会产生 AWS 账户费用（包括运行测试）。
• 建议为代码授予最低权限（least privilege），只授予完成任务所需的最小权限集。
• 该示例并未在所有 AWS 区域测试，请在支持相关服务的区域运行。

环境准备

运行本示例前，请先满足 Python 目录 README 中的标准前置条件（已配置 AWS 凭证的 Boto3 环境等）。此外还需要以下 Python 依赖（见 requirements.txt）：

• Flask 2.2.0 或更高版本
• Flask-Cors 3.0.10 或更高版本
• webargs 8.2.0 或更高版本
• boto3 1.26.81 或更高版本
• pytest 7.2.1 或更高版本（运行测试时需要）

在虚拟环境中执行以下命令即可安装全部依赖：

python -m pip install -r requirements.txt

创建 AWS 资源

Aurora Serverless 数据库集群与 Secrets Manager 密钥

示例要求 Aurora 集群开启Data API特性。截至 2024 年 3 月，可用的组合包括：

• 使用 Aurora Serverless v2 或集群内预置实例的 Aurora PostgreSQL；
• 使用 Serverless v1 集群的 Aurora MySQL 或 Aurora PostgreSQL。

本示例假定集群采用 Aurora PostgreSQL + Aurora Serverless v2 的组合，数据库凭据存放于 Secrets Manager 密钥中。

推荐使用 Aurora Serverless 应用资源 README 中提供的方式，通过AWS CDK或AWS CLI创建和管理资源。

方式一：AWS CDK 部署（在 resources/cdk/aurora_serverless_app 目录下，该示例基于 AWS CDK 2.132.1 构建和测试）：

npm install cdk deploy

部署完成后，栈会输出以下关键值（供后续配置使用）：

Outputs: doc-example-aurora-app.ClusterArn = arn:aws:rds:us-west-2:0123456789012:cluster:doc-example-aurora-app-docexampleauroraappcluster-1bqmf5EXAMPLE doc-example-aurora-app.DbName = auroraappdb doc-example-aurora-app.SecretArn = arn:aws:secretsmanager:us-west-2:0123456789012:secret:docexampleauroraappsecret8B-rEHdtEXAMPLE-111222

从 setup.ts 可以看到栈的核心配置：创建用户名docexampleadmin的 Secrets Manager 密钥、创建数据库auroraappdb，集群使用 Aurora PostgreSQL 15.5 引擎并开启enableDataApi: true，Serverless v2 容量范围 0.5～8 ACU，同时通过CfnOutput导出SecretArn、ClusterArn、DbName三个栈输出。

方式二：AWS CLI 部署：

aws cloudformation create-stack --template-body file://setup.yaml --stack-name YOUR_STACK_NAME

栈名称在区域内和账户内必须唯一（最多 128 字符，可含数字和连字符）。部署期间可用aws cloudformation describe-stacks --stack-name YOUR_STACK_NAME查看状态，当StackStatus变为CREATE_COMPLETE即就绪，再通过以下命令获取输出值：

aws cloudformation describe-stacks --stack-name STACK_NAME --query Stacks[0].Outputs --output text

输出形如：

SecretArn arn:aws:secretsmanager:us-west-2:0123456789012:secret:docexampleauroraappsecret8B-6N2njEXAMPLE-111222 ClusterArn arn:aws:rds:us-west-2:0123456789012:cluster:aurora-test-stack-docexampleauroraappcluster12345-kh39pEXAMPLE DbName auroraappdb

注意：Aurora Serverless 集群可能因闲置而进入暂停状态，首次执行语句时若收到包含 communications link failure 的BadRequestException，稍等片刻待集群唤醒后重试即可（这一点在代码中也有对应处理，见下文 storage.py 分析）。

创建 work_items 表

资源创建完成后，还需要在数据库中建立work_items表。可用 AWS CLI 或 AWS 管理控制台完成。

使用 AWS CLI

运行以下命令前，请把三个占位值替换为 CloudFormation 脚本的输出：

• CLUSTER_ARN：Aurora DB 集群 ARN，例如arn:aws:rds:us-west-2:123456789012:cluster:doc-example-aurora-app-docexampleauroraappcluster-15xfvaEXAMPLE
• SECRET_ARN：数据库凭据密钥 ARN，例如arn:aws:secretsmanager:us-west-2:123456789012:secret:docexampleauroraappsecret8B-xI1R8EXAMPLE-hfDaaj
• DATABASE：数据库名称，例如auroraappdb

aws rds-data execute-statement \ --resource-arn "CLUSTER_ARN" \ --database "DATABASE" \ --secret-arn "SECRET_ARN" \ --sql "create table work_items (iditem SERIAL PRIMARY KEY, description TEXT, guide VARCHAR(45), status TEXT, username VARCHAR(45), archived BOOL DEFAULT false);"

提示：\是 Linux 或 Mac 命令提示符的续行符，其他平台请替换为对应平台的续行符。

使用 AWS 管理控制台

1. 打开 Amazon RDS 控制台，选择Query Editor。
2. 在Database instance or cluster中选择你的数据库实例；如果使用 CloudFormation 脚本创建，名称以doc-example-aurora-app-开头。
3. 在Database username中选择Connect with a Secrets Manager ARN。
4. 输入包含数据库凭据的密钥 ARN。
5. 在Enter the name of the database or schema中输入数据库名（如auroraappdb）。
6. 选择Connect to database，即可在 SQL 查询控制台执行语句。

PostgreSQL 兼容数据库建表语句：

create table work_items ( iditem SERIAL PRIMARY KEY, description TEXT, guide VARCHAR(45), status TEXT, username VARCHAR(45), archived BOOL DEFAULT false );

MySQL 兼容数据库建表语句：

create table work_items ( iditem INT AUTO_INCREMENT PRIMARY KEY, description TEXT, guide VARCHAR(45), status TEXT, username VARCHAR(45), archived BOOL DEFAULT 0 );

验证 SES 发件邮箱

要向收件人发送邮件报告，至少需要有一个已在 Amazon SES 中验证的邮箱地址，该地址将作为报告的发送方。验证步骤：

1. 浏览器打开 Amazon SES 控制台。
2. 如有需要，选择你的 AWS 区域。
3. 选择Verified identities→Create identity。
4. 选择Email address，输入你拥有的邮箱地址，然后选择Create identity。
5. 你会收到来自 Amazon Web Services 的验证邮件，按其指引完成验证。

提示：本示例中发件方与收件方可以使用同一个邮箱账号。

运行示例

配置服务

运行服务前，需要在 config.py 中填写 AWS 资源值与验证邮箱。该文件默认内容如下：

CLUSTER_ARN = "NEED-CLUSTER-ARN" SECRET_ARN = "NEED-SECRET-ARN" DATABASE = "auroraappdb" TABLE_NAME = "work_items" SENDER_EMAIL = "NEED-SENDER-EMAIL" SECRET_KEY = "change-for-production!"

各配置项说明：

• CLUSTER_ARN：Aurora DB 集群 ARN，例如arn:aws:rds:us-west-2:123456789012:cluster:doc-example-aurora-app-docexampleauroraappcluster-15xfvaEXAMPLE
• SECRET_ARN：数据库凭据密钥 ARN，例如arn:aws:secretsmanager:us-west-2:123456789012:secret:docexampleauroraappsecret8B-xI1R8EXAMPLE-hfDaaj
• DATABASE：数据库名称，如auroraappdb
• TABLE_NAME：工作项表名，如work_items
• SENDER_EMAIL：在 Amazon SES 中注册的发送邮箱
• SECRET_KEY：Flask 会话使用的密钥，生产环境必须替换为真正的机密值

从 app.py 的源码可以看到，应用启动时会通过app.config.from_pyfile("config.py", silent=True)读取配置；如果CLUSTER_ARN、SECRET_ARN、DATABASE、TABLE_NAME仍为NEED-*占位值，会直接抛出RuntimeError提醒你先完成配置。

启动 REST 服务

本示例使用 Flask 托管本地 Web 服务器与 REST 服务。在命令提示符中运行：

flask --debug run -p 8080

• --debug：开发阶段输出更详细的日志；
• -p 8080：指定端口为 8080，以配合 Elwing 客户端使用。

服务运行后，即可向各端点发送 HTTP 请求来增、查、改工作项并发送邮件报告。

启动 Elwing 并选择 Item Tracker

1. 按照 Elwing README 的指引运行 Elwing：安装 NodeJS 18，在 resources/clients/react/elwing 目录执行npm i后运行npm start。
2. Elwing 启动后会自动打开浏览器并访问 http://localhost:3000/。
3. 在左侧导航栏选择Item Tracker插件。插件定义见 item-tracker 插件入口，其左侧导航中还内置了Verify SES email identity与Running Aurora queries两个外部帮助链接。

选择插件后，Elwing 会向 REST 服务发送请求获取现有活动工作项：

GET http://localhost:8080/api/items?archived=false

初始时表格为空，界面如下图所示。

选择Add item，填写各项值后选择Add即可新增工作项：

此时会向 REST 服务发送携带 JSON 负载的 POST 请求：

POST http://localhost:8080/api/items {"name":"Me", "guide":"python", "description":"Show how to add an item", "status":"In progress", "archived":false}

添加后，工作项会显示在表格中；选择某一行旁的Archive按钮可归档活动工作项：

这对应一个指定条目 ID 与archive动作的 PUT 请求：

PUT http://localhost:8080/api/items/8db8aaa4-6f04-4467-bd60-EXAMPLEGUID:archive

在右侧下拉列表中选择过滤器（如Archived），可只显示指定状态的工作项：

这对应携带archived查询参数的 GET 请求：

GET http://localhost:8080/api/items?archived=true

输入收件人邮箱并选择Send report，即可发送活动工作项的邮件报告：

这对应携带report动作的 POST 请求：

POST http://localhost:8080/api/items:report

注意：当 Amazon SES 账户处于沙箱（sandbox）状态时，发件方与收件方邮箱都必须已在 SES 中注册验证。

深入理解示例实现

应用工厂与路由设计

app.py 通过create_app(test_config=None)应用工厂创建 Flask 应用，其职责包括：加载配置、创建 Boto3 客户端、实例化存储与报告对象、注册 URL 路由。其中 Boto3 客户端创建逻辑为：测试模式下从test_config读取RDSDATA_CLIENT与SES_CLIENT，否则分别通过boto3.client("rds-data")和boto3.client("ses")创建（app.py#L75-L80）。

路由采用 Flask 的MethodView类来组织，例如下面这条路由把GET /api/items请求映射到ItemList.get方法（app.py#L84-L92）：

item_list_view = ItemList.as_view('item_list_api', storage) app.add_url_rule( '/api/items', defaults={'iditem': None}, view_func=item_list_view, methods=['GET'], strict_slashes=False)

完整的端点注册汇总：

注意源码中还通过CORS(app)抑制开发期与 React 联调时的跨域错误，并在注释中明确提示：部署应用时应移除这行。

REST 方法与参数校验、字段转换

HTTP 请求被路由到 ItemList 与 Report 类中的方法，二者使用webargs与marshmallow处理参数解析与数据转换。

web 页面中的字段名与数据库表字段名不一致，例如页面里叫id、数据表里叫iditem，页面里叫name、表里叫username。通过定义data_key，marshmallow schema 会自动完成字段名转换（item_list.py#L23-L34）：

class WorkItemSchema(Schema): iditem = fields.Str(data_key="id") description = fields.Str() guide = fields.Str() status = fields.Str() username = fields.Str(data_key="name") archived = fields.Bool()

ItemList中的各方法使用@use_args与@use_kwargs装饰器解析入参。例如get方法用@use_kwargs(WorkItemSchema, location="query")把查询字符串中的字段解析进方法签名，再调用底层storage对象获取工作项：

@use_kwargs(WorkItemSchema, location="query") def get(self, iditem, archived=None): work_items = self.storage.get_work_items(archived)

post方法则用@use_args(WorkItemSchema)校验并转换请求体 JSON。put方法只接受archive动作，其他动作一律返回 HTTP 400 与Unrecognized action ...错误信息（item_list.py#L103-L121）。

各方法还统一处理了两类异常并映射到对应 HTTP 状态码：

• DataServiceNotReadyException（Data Service 未就绪，常见于 Serverless 集群暂停后唤醒中）→ HTTP 503，提示"Wait a minute and try again"；
• StorageError（其他存储错误）→ HTTP 500。

Aurora Serverless 存储层：RDS Data Service 封装

storage.py 通过 Boto3 的rds-data客户端封装对 Aurora Serverless 数据库的读写。_run_statement把数据库名、集群 ARN、密钥 ARN 与 SQL 语句打包后调用execute_statement（storage.py#L47-L80）：

def _run_statement(self, sql, sql_params=None): run_args = { 'database': self._db_name, 'resourceArn': self._cluster, 'secretArn': self._secret, 'sql': sql } if sql_params is not None: run_args['parameters'] = sql_params results = self._rdsdata_client.execute_statement(**run_args)

该方法还针对 Aurora Serverless 的典型场景做了容错：当收到BadRequestException且错误信息包含 "Communications link failure" 时，抛出DataServiceNotReadyException，提示集群可能因闲置进入暂停模式，稍等约一分钟后重试。

get_work_items根据archived参数动态拼接WHERE archived=:archived与命名参数（布尔值通过booleanValue传递），然后执行SELECT iditem, description, guide, status, username, archived FROM ...，并把返回的records（RDS Data API 的行列式结果）转换为字典列表（storage.py#L82-L110）：

def get_work_items(self, archived=None): if archived is not None: sql_where = "WHERE archived=:archived" sql_params = [{'name': 'archived', 'value': {'booleanValue': archived}}] sql = f"SELECT iditem, description, guide, status, username, archived FROM {self._table_name} {sql_where}" results = self._run_statement(sql, sql_params=sql_params)

add_work_item的实现细节尤其值得注意（storage.py#L112-L148）：由于 Aurora Serverless v2 的 Data API 对 DML 语句返回的generatedFields全部为空，代码把 INSERT 包装进WITH子句并附带RETURNING iditem，再以查询方式取回自增 ID：

sql = ( "WITH t1 AS ( " f"INSERT INTO {self._table_name} (description, guide, status, username) " " VALUES (:description, :guide, :status, :username) RETURNING iditem " ") SELECT iditem FROM t1" )

注释中同时保留了旧版（Serverless v1）从generatedFields[0]["longValue"]读取 ID 的写法，并说明该限制可能并非永久、后续 DML 语句可能被简化。archive_work_item则通过命名参数archived: True（booleanValue）与iditem（longValue）执行UPDATE ... SET archived=:archived WHERE iditem=:iditem。

Amazon SES 邮件报告：按数据量自适应发送策略

report.py 负责把工作项渲染成报告并发送邮件，其核心逻辑在Report.post（report.py#L89-L158）：

1. 先从存储层取出所有活动工作项（archived=False）；
2. 通过render_template渲染 HTML 版报告（模板见 templates/report.html）与文本版报告（模板见 templates/report.txt），模板内包含条目数、快照时间以及逐条工作项表格；同时用_render_csv把工作项渲染为 CSV 字符串；
3. 当报告包含的工作项≤ 10 条时，直接使用 SES 的send_email动作发送，HTML 与文本正文以纯 Python 字符串传入：

   self.ses_client.send_email( Source=self.email_sender, Destination={"ToAddresses": [email]}, Message={ "Subject": {"Data": "Work items"}, "Body": { "Html": {"Data": html_report}, "Text": {"Data": text_report}, }, }, )

4. 当工作项超过 10 条时，CSV 作为附件、邮件必须以 MIME 格式发送，因此改用send_raw_email动作：_format_mime_message构造MIMEMultipart("mixed")消息，主体再嵌套MIMEMultipart("alternative")承载 text/html 两个版本，并附加文件名work_items.csv的MIMEApplication附件（report.py#L48-L68）：

   mime_msg = self._format_mime_message(email, text_report, html_report, csv_items) response = self.ses_client.send_raw_email( Source=self.email_sender, Destinations=[email], RawMessage={"Data": mime_msg.as_string()}, )

Report.post通过@use_kwargs({"email": fields.Str(required=True)})强制要求请求体携带email字段。异常处理同样细致：存储层错误返回 HTTP 500 并提示 "A storage error occurred."，SESClientError返回 HTTP 500 并提示 "An email error occurred."。

测试验证：行为与错误路径全覆盖

示例自带基于 pytest + botocore stubber 的单元测试，见 test/test_app.py，测试夹具定义在 test/conftest.py。测试通过create_app注入TESTING、RDSDATA_CLIENT、SES_CLIENT等配置来构造测试应用，再借助make_stubber与stub_runner拦截execute_statement、send_email、send_raw_email等 API 调用。测试用例覆盖：

• GET /api/items在archived=false、archived=true、不传参三种情况下的正确返回；
• GET与POST遇到BadRequestException（Data Service 未就绪）时返回 503、遇到存储异常时返回 500；
• POST /api/items成功新增并返回生成的 ID；
• PUT /api/items/<id>:archive成功归档，未知动作（如garbage）返回 400 与Unrecognized action提示；
• POST /api/items:report在报告条数较少（≤10）时走send_email、条数较多时走send_raw_email两条路径，以及存储错误 / 邮件发送错误场景。

可以在 python/cross_service/aurora_item_tracker 目录运行pytest执行测试。

清理资源

为避免持续计费，请删除教程中创建的所有资源：

• 若使用 AWS CDK 或 AWS CLI 创建资源，按 Aurora Serverless 应用资源 README 的指引销毁即可：CDK 执行cdk destroy，CLI 执行aws cloudformation delete-stack --stack-name YOUR_STACK_NAME。
• 若通过 AWS 管理控制台创建或运行应用时改动了资源，则需要通过控制台手动删除。

小结与下一步

至此，你已经完成了一个完整的跨服务示例：构建了读写、归档 Aurora Serverless 数据库中工作项的 REST 服务，通过 Secrets Manager 完成数据库认证，并使用 Amazon SES 向注册用户发送邮件报告。核心文件速查：

• app.py：应用工厂、Boto3 客户端创建、URL 路由注册；
• config.py：服务运行配置；
• item_list.py：REST 资源方法、webargs/marshmallow 参数解析与字段转换；
• storage.py：RDS Data Service 封装与 SQL 执行；
• report.py：HTML/文本/CSV 报告渲染与 SES 发送；
• test/test_app.py：端到端行为与错误路径测试。

延伸阅读（官方文档）：Amazon Aurora 用户指南、Amazon RDS 用户指南、Amazon SES 开发者指南、Amazon RDS Data Service Boto3 API 参考、Amazon SES Boto3 API 参考。若想进一步探索，可在本仓库中找到同一模式的 DynamoDB 版 Item Tracker（python/cross_service/dynamodb_item_tracker）等兄弟示例作为对比参考。

• 示例工程
• 教程
• 后端