企业官网建设流程全解析

• 示例工程
• 教程
• 后端

本文以 aws-doc-sdk-examples 仓库中python/example_code/cloudfront目录下的官方示例为蓝本，系统讲解如何用 AWS SDK for Python（Boto3）执行 Amazon CloudFront 分发（Distribution）的查询与更新操作。读完本文，你将掌握ListDistributions、GetDistributionConfig、UpdateDistribution三个核心 API 的调用方式、ETag 乐观锁的更新流程，以及如何借助仓库自带的 Stubber 测试工具完成不产生任何 AWS 费用的单元测试。

示例概览

本示例围绕 CloudFront 的"分发"这一核心资源展开，封装在CloudFrontWrapper类中，位于 distributions.py。它演示了三类最常见的单动作（Single action）调用：

CloudFront 通过全球边缘节点加速静态与动态 Web 内容（如.html、.css、.php、图片和媒体文件）的分发，而"分发"正是定义这些内容如何被缓存与交付的配置单元。本示例即围绕读取、修改这一配置单元展开。

环境准备与运行方式

前置条件

• 一个有效的 AWS 账户，并按 AWS 工具与 SDK 共享配置和凭据参考指南 中有完整说明）。
• Python 3.6.0 或更高版本。
• 遵循最低权限原则：仅授予执行任务所需的最小 IAM 权限。对于本示例，即 CloudFront 的ListDistributions、GetDistributionConfig、UpdateDistribution三个操作的读/写权限。

安装依赖

在虚拟环境中安装 requirements.txt 列出的包：

python -m pip install -r requirements.txt

该文件声明了两个依赖：

• boto3>=1.26.79：AWS SDK for Python，提供 CloudFront 客户端；
• pytest>=7.2.1：运行单元测试所需的测试框架。

推荐的完整流程是先创建并激活虚拟环境：

python -m venv .venv source .venv/bin/activate # Linux、macOS 或 Unix；Windows 使用 .venv\Scripts\activate python -m pip install -r requirements.txt

运行示例

在命令行中直接运行：

python distributions.py

脚本会依次执行两个操作：

1. 调用list_distributions()打印账户下的所有 CloudFront 分发；
2. 调用update_distribution()交互式地等待你输入一个分发 ID，然后修改该分发的 Comment。

⚠ 注意：在真实账户上运行会产生 AWS 费用。若仅做学习验证，建议优先使用下文介绍的 Stubber 单元测试方式。

核心代码逐段解析

客户端封装与类结构

示例用一个薄封装类持有 Boto3 CloudFront 客户端：

class CloudFrontWrapper: """Encapsulates Amazon CloudFront operations.""" def __init__(self, cloudfront_client): """ :param cloudfront_client: A Boto3 CloudFront client """ self.cloudfront_client = cloudfront_client

构造函数接收一个boto3.client("cloudfront")实例，把"创建客户端"与"业务操作"解耦——这正是单元测试中注入 Stubber 的关键：测试替身替换的是这里的客户端对象，而非业务逻辑。

ListDistributions：枚举分发列表

def list_distributions(self): print("CloudFront distributions:\n") distributions = self.cloudfront_client.list_distributions() if distributions["DistributionList"]["Quantity"] > 0: for distribution in distributions["DistributionList"]["Items"]: print(f"Domain: {distribution['DomainName']}") print(f"Distribution Id: {distribution['Id']}") print( f"Certificate Source: " f"{distribution['ViewerCertificate']['CertificateSource']}" ) if distribution["ViewerCertificate"]["CertificateSource"] == "acm": print( f"Certificate: {distribution['ViewerCertificate']['Certificate']}" ) print("") else: print("No CloudFront distributions detected.")

实现要点：

• 分页结构：list_distributions()的返回体DistributionList是 CloudFront 统一的"量 + 项"结构。Quantity表示条目总数，Items是实际数据数组。代码先判断Quantity > 0再遍历，避免空列表报错。
• 证书来源判断：ViewerCertificate.CertificateSource字段取值通常为acm（AWS Certificate Manager）、iam（IAM 证书存储，已不再推荐）或cloudfront（默认证书）。示例仅当来源为acm时才打印证书 ARN，因为只有该来源的Certificate字段才有意义。
• 每条分发还包含Status（如Deployed）、PriceClass、Enabled、HttpVersion等字段，详见下方测试桩的定义。

GetDistributionConfig + UpdateDistribution：带并发保护的更新

def update_distribution(self): distribution_id = input( "This script updates the comment for a CloudFront distribution.\n" "Enter a CloudFront distribution ID: " ) distribution_config_response = self.cloudfront_client.get_distribution_config( Id=distribution_id ) distribution_config = distribution_config_response["DistributionConfig"] distribution_etag = distribution_config_response["ETag"] distribution_config["Comment"] = input( f"\nThe current comment for distribution {distribution_id} is " f"'{distribution_config['Comment']}'.\n" f"Enter a new comment: " ) self.cloudfront_client.update_distribution( DistributionConfig=distribution_config, Id=distribution_id, IfMatch=distribution_etag, ) print("Done!")

这是本示例最有价值的部分，完整演示了 CloudFront 分发更新的**读-改-写（read-modify-write）**流程：

1. 读取现有配置：get_distribution_config(Id=...)返回完整的DistributionConfig结构体，以及与之配套的ETag。CloudFront 分发的配置是强类型、全量提交的——update_distribution不接受部分字段补丁，而是要求提交完整的DistributionConfig。
2. 本地修改：仅对目标字段（Comment）做局部改动，其余字段原样保留，避免覆盖丢失其他配置。
3. 乐观并发控制：update_distribution的IfMatch参数携带先前拿到的 ETag。CloudFront 会比较该值与服务端当前版本：若在此期间分发被其他操作修改（ETag 变化），请求会被拒绝并返回PreconditionFailed（412），防止"最后写入覆盖先写"的竞态问题。这正是 ETag 机制的核心价值——它保证了并发更新下的数据一致性。

需要注意，更新分发是异步生效的：请求成功后分发进入InProgress状态，配置变更需要一定时间才会在全域边缘节点完成部署。

单元测试：用 Stubber 零成本验证行为

测试原理

仓库为 CloudFront 准备了专属测试桩 cloudfront_stubber.py，其设计基于 botocore Stubber：在单元测试模式下，stub_list_distributions、stub_get_distribution_config、stub_update_distribution会拦截 Boto3 客户端调用并返回预设响应，请求不发送到 AWS，也不会产生任何费用；而当通过USE_AWS等机制切换为真实调用时，桩函数自动透传（passthrough）。这一开关由 common.py 中的make_stubberfixture 统一管理。

测试用例解读

测试文件位于 test/test_distributions.py，使用 conftest.py 通过sys.path.append("../..")引入公共测试工具。

test_list_distributions用@pytest.mark.parametrize对error_code参数化（None与"TestException"两种场景），分别验证正常路径与异常路径：

@pytest.mark.parametrize("error_code", [None, "TestException"]) def test_list_distributions(make_stubber, error_code): cloudfront_client = boto3.client("cloudfront") cloudfront_stubber = make_stubber(cloudfront_client) cloudfront = CloudFrontWrapper(cloudfront_client) distribs = [ { "name": f"distrib-name-{index}", "id": f"distrib-{index}", "cert_source": "acm", "cert": "Hi, I'm a certificate!", } for index in range(3) ] cloudfront_stubber.stub_list_distributions(distribs, error_code=error_code) if error_code is None: cloudfront.list_distributions() else: with pytest.raises(ClientError) as exc_info: cloudfront.list_distributions() assert exc_info.value.response["Error"]["Code"] == error_code

test_update_distribution用monkeypatch替换builtins.input，把交互输入替换为预设的["test-id", comment]，从而在无人工干预下走完整条读-改-写链路：

inputs = ["test-id", comment] monkeypatch.setattr("builtins.input", lambda x: inputs.pop(0)) cloudfront_stubber.stub_get_distribution_config(distrib_id, comment, etag) cloudfront_stubber.stub_update_distribution( distrib_id, comment, etag, error_code=error_code )

测试桩的预期参数校验

cloudfront_stubber.py 中每个 stub 函数都定义了expected_params，Stubber 会校验实际请求参数是否与预期一致：

• stub_list_distributions：构造含DistributionList的完整响应，其中每个Items条目覆盖了ARN、Status、Origins、DefaultCacheBehavior、PriceClass、Enabled、Restrictions、ViewerCertificate、Staging等字段，可见一个分发对象实际包含的配置维度；
• stub_get_distribution_config：校验请求参数{"Id": distrib_id}，返回DistributionConfig（含CallerReference、Origins、DefaultCacheBehavior、Enabled、Comment）与ETag；
• stub_update_distribution：校验请求参数必须同时包含Id、完整的DistributionConfig和IfMatch=etag——这从测试层印证了"更新必须全量配置 + ETag 并发保护"的 API 契约。

运行测试

在 python/README.md 的测试章节中，单元测试统一通过排除integ标记运行：

python -m pytest -m "not integ"

在示例目录下执行即可验证全部单元测试。集成测试（真实 AWS 调用）则通过包含integ标记运行，会创建/销毁真实资源并产生费用，需谨慎操作。

关键设计要点总结

1. 全量配置提交：CloudFront 的update_distribution要求提交完整DistributionConfig，因此正确的模式永远是"先get_distribution_config取出、局部修改、再整体提交"，绝不能凭记忆手工构造配置字典。
2. ETag 乐观锁：IfMatch参数携带的 ETag 是并发安全的基石；更新前必须保留get_distribution_config返回的 ETag，否则在并发场景下可能发生静默覆盖。
3. 客户端注入便于测试：CloudFrontWrapper接受外部传入客户端，配合仓库统一的 Stubber 机制，可以在不触碰 AWS 的情况下覆盖正常与异常两条路径。
4. 交互式脚本：示例通过input()驱动流程，直接运行时需要人工输入；自动化测试则用monkeypatch模拟输入，验证同样的逻辑。

延伸阅读

• 本示例依赖的通用测试基础设施：python/test_tools/fixtures/common.py（make_stubberfixture 与错误注入机制）与 python/test_tools/stubber_factory.py
• Python 全量示例的安装、运行、测试与 Docker 容器说明：python/README.md
• CloudFront 分发的完整字段定义可在 Boto3 的cloudfront服务参考中找到（Distribution、DistributionConfig、ViewerCertificate等类型），建议在编码时结合查阅。

本文基于 aws-doc-sdk-examples 仓库中的 cloudfront README 及其配套源码编写，示例代码遵循 Apache-2.0 许可。运行示例与测试前请务必确认你的 AWS 凭据配置正确，并了解可能产生的账户费用。

• 示例工程
• 教程
• 后端