使用 Boto3 管理 CloudFront 分发：Python 代码示例实战指南

示例工程教程后端【免费下载链接】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-doc-sdk-examples 仓库中python/example_code/cloudfront目录下的官方示例为蓝本系统讲解如何用 AWS SDK for PythonBoto3执行 Amazon CloudFront 分发Distribution的查询与更新操作。读完本文你将掌握ListDistributions、GetDistributionConfig、UpdateDistribution三个核心 API 的调用方式、ETag 乐观锁的更新流程以及如何借助仓库自带的 Stubber 测试工具完成不产生任何 AWS 费用的单元测试。示例概览本示例围绕 CloudFront 的分发这一核心资源展开封装在CloudFrontWrapper类中位于 distributions.py。它演示了三类最常见的单动作Single action调用单动作源码位置对应 Boto3 方法功能ListDistributionsdistributions.py#L27list_distributions()列举账户下的全部分发并打印域名、ID 与证书信息GetDistributionConfigdistributions.py#L55get_distribution_config()获取指定分发的配置与 ETag作为更新示例的预备步骤UpdateDistributiondistributions.py#L49update_distribution()在并发保护下更新分发的 Comment 字段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该文件声明了两个依赖boto31.26.79AWS SDK for Python提供 CloudFront 客户端pytest7.2.1运行单元测试所需的测试框架。推荐的完整流程是先创建并激活虚拟环境python -m venv .venv source .venv/bin/activate # Linux、macOS 或 UnixWindows 使用 .venv\Scripts\activate python -m pip install -r requirements.txt运行示例在命令行中直接运行python distributions.py脚本会依次执行两个操作调用list_distributions()打印账户下的所有 CloudFront 分发调用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(fDomain: {distribution[DomainName]}) print(fDistribution Id: {distribution[Id]}) print( fCertificate Source: f{distribution[ViewerCertificate][CertificateSource]} ) if distribution[ViewerCertificate][CertificateSource] acm: print( fCertificate: {distribution[ViewerCertificate][Certificate]} ) print() else: print(No CloudFront distributions detected.)实现要点分页结构list_distributions()的返回体DistributionList是 CloudFront 统一的量 项结构。Quantity表示条目总数Items是实际数据数组。代码先判断Quantity 0再遍历避免空列表报错。证书来源判断ViewerCertificate.CertificateSource字段取值通常为acmAWS Certificate Manager、iamIAM 证书存储已不再推荐或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( Iddistribution_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 fEnter a new comment: ) self.cloudfront_client.update_distribution( DistributionConfigdistribution_config, Iddistribution_id, IfMatchdistribution_etag, ) print(Done!)这是本示例最有价值的部分完整演示了 CloudFront 分发更新的**读-改-写read-modify-write**流程读取现有配置get_distribution_config(Id...)返回完整的DistributionConfig结构体以及与之配套的ETag。CloudFront 分发的配置是强类型、全量提交的——update_distribution不接受部分字段补丁而是要求提交完整的DistributionConfig。本地修改仅对目标字段Comment做局部改动其余字段原样保留避免覆盖丢失其他配置。乐观并发控制update_distribution的IfMatch参数携带先前拿到的 ETag。CloudFront 会比较该值与服务端当前版本若在此期间分发被其他操作修改ETag 变化请求会被拒绝并返回PreconditionFailed412防止最后写入覆盖先写的竞态问题。这正是 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: fdistrib-name-{index}, id: fdistrib-{index}, cert_source: acm, cert: Hi, Im a certificate!, } for index in range(3) ] cloudfront_stubber.stub_list_distributions(distribs, error_codeerror_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_codetest_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_codeerror_code )测试桩的预期参数校验cloudfront_stubber.py 中每个 stub 函数都定义了expected_paramsStubber 会校验实际请求参数是否与预期一致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与ETagstub_update_distribution校验请求参数必须同时包含Id、完整的DistributionConfig和IfMatchetag——这从测试层印证了更新必须全量配置 ETag 并发保护的 API 契约。运行测试在 python/README.md 的测试章节中单元测试统一通过排除integ标记运行python -m pytest -m not integ在示例目录下执行即可验证全部单元测试。集成测试真实 AWS 调用则通过包含integ标记运行会创建/销毁真实资源并产生费用需谨慎操作。关键设计要点总结全量配置提交CloudFront 的update_distribution要求提交完整DistributionConfig因此正确的模式永远是先get_distribution_config取出、局部修改、再整体提交绝不能凭记忆手工构造配置字典。ETag 乐观锁IfMatch参数携带的 ETag 是并发安全的基石更新前必须保留get_distribution_config返回的 ETag否则在并发场景下可能发生静默覆盖。客户端注入便于测试CloudFrontWrapper接受外部传入客户端配合仓库统一的 Stubber 机制可以在不触碰 AWS 的情况下覆盖正常与异常两条路径。交互式脚本示例通过input()驱动流程直接运行时需要人工输入自动化测试则用monkeypatch模拟输入验证同样的逻辑。延伸阅读本示例依赖的通用测试基础设施python/test_tools/fixtures/common.pymake_stubberfixture 与错误注入机制与 python/test_tools/stubber_factory.pyPython 全量示例的安装、运行、测试与 Docker 容器说明python/README.mdCloudFront 分发的完整字段定义可在 Boto3 的cloudfront服务参考中找到Distribution、DistributionConfig、ViewerCertificate等类型建议在编码时结合查阅。本文基于 aws-doc-sdk-examples 仓库中的 cloudfront README 及其配套源码编写示例代码遵循 Apache-2.0 许可。运行示例与测试前请务必确认你的 AWS 凭据配置正确并了解可能产生的账户费用。赞分享示例工程教程后端【免费下载链接】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 SDK for PythonBoto3管理 Amazon EC2示例代码全指南使用 AWS SDK for PythonBoto3管理 Amazon EC2示例代码全指南 本文围绕 AWS Code Examples 仓库中 pyt示例工程教程后端AWS SDK for Python (Boto3) 代码示例详解与使用指南AWS SDK for Python Boto3 代码示例详解与使用指南 前言 AWS SDK for PythonBoto3是AWS官方提供的Python示例工程教程后端IoT-For-Beginners 实战用 Raspberry Pi 与 Grove 继电器实现基于土壤湿度的自动浇水IoT For Beginners 实战用 Raspberry Pi 与 Grove 继电器实现基于土壤湿度的自动浇水 本篇技术指南聚焦于 Microsoft示例工程教程后端上一篇炉石传说HsMod插件55项功能免费增强你的游戏体验下一篇3分钟上手MASTG合规检查工具从安装到实战的安全测试加速指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考