示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载导读本文以 AWS 官方代码示例仓库中的 Aurora Item Tracker 项目为主线讲解如何用 AWS SDK for PythonBoto3搭建一个基于 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 Servicerds-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.txtFlask 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: trueServerless v2 容量范围 0.58 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_ARNAurora DB 集群 ARN例如arn:aws:rds:us-west-2:123456789012:cluster:doc-example-aurora-app-docexampleauroraappcluster-15xfvaEXAMPLESECRET_ARN数据库凭据密钥 ARN例如arn:aws:secretsmanager:us-west-2:123456789012:secret:docexampleauroraappsecret8B-xI1R8EXAMPLE-hfDaajDATABASE数据库名称例如auroraappdbaws 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 管理控制台打开 Amazon RDS 控制台选择Query Editor。在Database instance or cluster中选择你的数据库实例如果使用 CloudFormation 脚本创建名称以doc-example-aurora-app-开头。在Database username中选择Connect with a Secrets Manager ARN。输入包含数据库凭据的密钥 ARN。在Enter the name of the database or schema中输入数据库名如auroraappdb。选择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 中验证的邮箱地址该地址将作为报告的发送方。验证步骤浏览器打开 Amazon SES 控制台。如有需要选择你的 AWS 区域。选择Verified identities→Create identity。选择Email address输入你拥有的邮箱地址然后选择Create identity。你会收到来自 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_ARNAurora DB 集群 ARN例如arn:aws:rds:us-west-2:123456789012:cluster:doc-example-aurora-app-docexampleauroraappcluster-15xfvaEXAMPLESECRET_ARN数据库凭据密钥 ARN例如arn:aws:secretsmanager:us-west-2:123456789012:secret:docexampleauroraappsecret8B-xI1R8EXAMPLE-hfDaajDATABASE数据库名称如auroraappdbTABLE_NAME工作项表名如work_itemsSENDER_EMAIL在 Amazon SES 中注册的发送邮箱SECRET_KEYFlask 会话使用的密钥生产环境必须替换为真正的机密值从 app.py 的源码可以看到应用启动时会通过app.config.from_pyfile(config.py, silentTrue)读取配置如果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按照 Elwing README 的指引运行 Elwing安装 NodeJS 18在 resources/clients/react/elwing 目录执行npm i后运行npm start。Elwing 启动后会自动打开浏览器并访问 http://localhost:3000/。在左侧导航栏选择Item Tracker插件。插件定义见 item-tracker 插件入口其左侧导航中还内置了Verify SES email identity与Running Aurora queries两个外部帮助链接。选择插件后Elwing 会向 REST 服务发送请求获取现有活动工作项GET http://localhost:8080/api/items?archivedfalse初始时表格为空界面如下图所示。选择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?archivedtrue输入收件人邮箱并选择Send report即可发送活动工作项的邮件报告这对应携带report动作的 POST 请求POST http://localhost:8080/api/items:report注意当 Amazon SES 账户处于沙箱sandbox状态时发件方与收件方邮箱都必须已在 SES 中注册验证。深入理解示例实现应用工厂与路由设计app.py 通过create_app(test_configNone)应用工厂创建 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-L92item_list_view ItemList.as_view(item_list_api, storage) app.add_url_rule( /api/items, defaults{iditem: None}, view_funcitem_list_view, methods[GET], strict_slashesFalse)完整的端点注册汇总HTTP 方法路由处理对象与方法作用GET/api/itemsItemList.get获取全部或按archived过滤的工作项POST/api/itemsItemList.post新增工作项GET / PUT/api/items/iditemItemList.get/ItemList.put获取单个工作项 / 更新PUT/api/items/iditem:actionItemList.put按动作仅archive更新条目POST/api/items:reportReport.post发送工作项邮件报告注意源码中还通过CORS(app)抑制开发期与 React 联调时的跨域错误并在注释中明确提示部署应用时应移除这行。REST 方法与参数校验、字段转换HTTP 请求被路由到 ItemList 与 Report 类中的方法二者使用webargs与marshmallow处理参数解析与数据转换。web 页面中的字段名与数据库表字段名不一致例如页面里叫id、数据表里叫iditem页面里叫name、表里叫username。通过定义data_keymarshmallow schema 会自动完成字段名转换item_list.py#L23-L34class WorkItemSchema(Schema): iditem fields.Str(data_keyid) description fields.Str() guide fields.Str() status fields.Str() username fields.Str(data_keyname) archived fields.Bool()ItemList中的各方法使用use_args与use_kwargs装饰器解析入参。例如get方法用use_kwargs(WorkItemSchema, locationquery)把查询字符串中的字段解析进方法签名再调用底层storage对象获取工作项use_kwargs(WorkItemSchema, locationquery) def get(self, iditem, archivedNone): 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 状态码DataServiceNotReadyExceptionData Service 未就绪常见于 Serverless 集群暂停后唤醒中→ HTTP 503提示Wait a minute and try againStorageError其他存储错误→ HTTP 500。Aurora Serverless 存储层RDS Data Service 封装storage.py 通过 Boto3 的rds-data客户端封装对 Aurora Serverless 数据库的读写。_run_statement把数据库名、集群 ARN、密钥 ARN 与 SQL 语句打包后调用execute_statementstorage.py#L47-L80def _run_statement(self, sql, sql_paramsNone): 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 ...并把返回的recordsRDS Data API 的行列式结果转换为字典列表storage.py#L82-L110def get_work_items(self, archivedNone): if archived is not None: sql_where WHERE archived:archived sql_params [{name: archived, value: {booleanValue: archived}}] sql fSELECT iditem, description, guide, status, username, archived FROM {self._table_name} {sql_where} results self._run_statement(sql, sql_paramssql_params)add_work_item的实现细节尤其值得注意storage.py#L112-L148由于 Aurora Serverless v2 的 Data API 对 DML 语句返回的generatedFields全部为空代码把 INSERT 包装进WITH子句并附带RETURNING iditem再以查询方式取回自增 IDsql ( WITH t1 AS ( fINSERT 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: TruebooleanValue与iditemlongValue执行UPDATE ... SET archived:archived WHERE iditem:iditem。Amazon SES 邮件报告按数据量自适应发送策略report.py 负责把工作项渲染成报告并发送邮件其核心逻辑在Report.postreport.py#L89-L158先从存储层取出所有活动工作项archivedFalse通过render_template渲染 HTML 版报告模板见 templates/report.html与文本版报告模板见 templates/report.txt模板内包含条目数、快照时间以及逐条工作项表格同时用_render_csv把工作项渲染为 CSV 字符串当报告包含的工作项≤ 10 条时直接使用 SES 的send_email动作发送HTML 与文本正文以纯 Python 字符串传入self.ses_client.send_email( Sourceself.email_sender, Destination{ToAddresses: [email]}, Message{ Subject: {Data: Work items}, Body: { Html: {Data: html_report}, Text: {Data: text_report}, }, }, )当工作项超过 10 条时CSV 作为附件、邮件必须以 MIME 格式发送因此改用send_raw_email动作_format_mime_message构造MIMEMultipart(mixed)消息主体再嵌套MIMEMultipart(alternative)承载 text/html 两个版本并附加文件名work_items.csv的MIMEApplication附件report.py#L48-L68mime_msg self._format_mime_message(email, text_report, html_report, csv_items) response self.ses_client.send_raw_email( Sourceself.email_sender, Destinations[email], RawMessage{Data: mime_msg.as_string()}, )Report.post通过use_kwargs({email: fields.Str(requiredTrue)})强制要求请求体携带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在archivedfalse、archivedtrue、不传参三种情况下的正确返回GET与POST遇到BadRequestExceptionData Service 未就绪时返回 503、遇到存储异常时返回 500POST /api/items成功新增并返回生成的 IDPUT /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 destroyCLI 执行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.pyREST 资源方法、webargs/marshmallow 参数解析与字段转换storage.pyRDS Data Service 封装与 SQL 执行report.pyHTML/文本/CSV 报告渲染与 SES 发送test/test_app.py端到端行为与错误路径测试。延伸阅读官方文档Amazon Aurora 用户指南、Amazon RDS 用户指南、Amazon SES 开发者指南、Amazon RDS Data Service Boto3 API 参考、Amazon SES Boto3 API 参考。若想进一步探索可在本仓库中找到同一模式的 DynamoDB 版 Item Trackerpython/cross_service/dynamodb_item_tracker等兄弟示例作为对比参考。赞分享示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载相关推荐使用 Boto3 构建 DynamoDB 工作项追踪器 REST 服务从 Flask 应用到 Amazon SES 邮件报告使用 Boto3 构建 DynamoDB 工作项追踪器 REST 服务从 Flask 应用到 Amazon SES 邮件报告 导读 本文基于 AWS 官方示例示例工程教程后端AWS SDK for JavaScript v3 跨服务实战用 Aurora Serverless 与 SES 构建工作项跟踪 REST 服务aws-doc-sdk-examplesAWS SDK for JavaScript v3 跨服务实战用 Aurora Serverless 与 SES 构建工作项跟踪 REST 服务aws do示例工程教程后端使用 AWS SDK for .NET (v3) 构建基于 Aurora Serverless 的工作项追踪 REST 服务使用 AWS SDK for .NET v3 构建基于 Aurora Serverless 的工作项追踪 REST 服务 导读 本篇文章围绕 dotnetv3/示例工程教程后端上一篇GTA圣安地列斯存档编辑器终极游戏进度掌控免费工具下一篇终极指南3步免费将VR视频转换为可交互的2D体验创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考