Python凭借其简洁的语法、丰富的生态以及跨平台特性，成为数据科学、云计算、自动化运维等领域的核心工具。从Web开发中Django框架的高效路由处理，到机器学习中Scikit-learn的模型构建，再到量化交易中Pandas的数据清洗，Python以其模块化设计实现了对复杂业务场景的灵活支撑。在云计算领域，Python同样扮演着关键角色，而AWS CLI（Amazon Web Services Command Line Interface）作为连接Python与AWS云服务的桥梁，为开发者提供了从命令行到脚本化操作的全链路解决方案。本文将深入解析AWS CLI的核心功能、工作机制及在Python脚本中的实战应用，助您快速掌握云资源自动化管理的核心技能。
一、AWS CLI核心功能与技术特性解析
1.1 工具定位与应用场景
AWS CLI是亚马逊官方提供的开源命令行工具，允许用户通过命令行或脚本直接操作AWS服务。其核心功能覆盖EC2实例管理、S3存储桶操作、RDS数据库配置、Lambda函数部署等100+项AWS服务，可满足从资源创建、状态查询到成本监控的全生命周期管理需求。典型应用场景包括：
- 批量资源管理：通过脚本批量创建EC2实例并配置安全组
- 持续集成/部署：在CI/CD流程中自动化部署Lambda函数
- 数据备份与同步：定期将本地数据同步至S3存储桶并启用版本控制
- 成本优化脚本：查询未使用的EBS卷并自动释放以节省成本
1.2 技术架构与工作原理
AWS CLI基于Python开发，底层通过Boto3库与AWS API进行交互。其工作流程如下：
- 用户输入命令（如
aws s3 ls） - CLI解析命令参数并生成对应的API请求
- 通过HTTP协议将请求发送至AWS服务端点
- 接收API响应并格式化输出结果
- 在脚本环境中可捕获输出结果供后续逻辑处理
该工具采用插件式架构，支持通过自定义插件扩展功能，例如通过awscli-plugin-endpoint插件配置私有API端点。
1.3 优势与局限性分析
核心优势：
- 功能完整性：覆盖几乎所有AWS服务的API操作
- 脚本友好性：输出结果支持JSON格式，便于脚本解析
- 版本兼容性：通过
aws --version命令可查看当前版本并支持版本升级 - 安全集成：无缝对接IAM角色与访问密钥（Access Key）体系
局限性：
- 学习成本：需掌握百余条命令的语法规则
- 性能瓶颈：批量操作时需处理API调用频率限制（Throttling）
- 依赖环境：需提前配置AWS凭证（credentials）与区域（region）
1.4 开源协议与合规性
AWS CLI遵循Apache License 2.0开源协议，允许用户自由使用、修改及分发，甚至可用于商业项目。企业在使用时需注意：
- 遵守AWS服务条款与数据合规要求
- 自定义插件需同样遵循Apache协议
- 涉及加密功能时需符合当地加密法规
二、多平台安装指南与环境配置
2.1 系统兼容性与安装方式
AWS CLI支持以下操作系统：
| 操作系统 | 推荐安装方式 | 依赖组件 |
|---|---|---|
| Windows | MSI安装包 / pip | Python 3.7+ |
| macOS | Homebrew / pip | Python 3.7+ / CLI Tools |
| Linux | apt-get / yum / pip | Python 3.7+ / GCC |
| Docker | 官方Docker镜像 | Docker Engine 20.10+ |
2.2 详细安装步骤（以macOS为例）
方式一：通过Homebrew安装（推荐）
# 更新Homebrew
brew update
# 安装AWS CLI v2（当前最新版本为2.13.15）
brew install awscli
# 验证安装
aws --version
# 预期输出：aws-cli/2.13.15 Python/3.12.0 Darwin/22.6.0 exe/x86_64 prompt/off
方式二：通过pip安装
# 安装Python包管理工具pip（若未安装）
sudo easy_install pip
# 安装AWS CLI
pip install awscli --upgrade --user
# 添加可执行文件路径到环境变量
echo 'export PATH="$PATH:~/.local/bin"' >> ~/.bash_profile
source ~/.bash_profile
2.3 环境配置与凭证管理
步骤1：配置默认区域与输出格式
aws configure
# 交互式配置：
# AWS Access Key ID [None]: AKIAXXXXXXXXXXXXXX
# AWS Secret Access Key [None]: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# Default region name [None]: us-west-2
# Default output format [None]: json
步骤2：多账户管理（可选）
在~/.aws/credentials文件中添加多个Profile：
[default]
aws_access_key_id = AKIAXXXXXXXXXXXXXX
aws_secret_access_key = xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
[dev-account]
aws_access_key_id = AKIAYYYYYYYYYYYYYY aws_secret_access_key = yyyyyyyyyyyyyyyyyyyyyyyyyyyyy region = eu-central-1
步骤3：临时安全凭证（适用于IAM角色）
# 假设已通过STS获取临时凭证
export AWS_ACCESS_KEY_ID=AKIAXXXXXXXXXXXXXX
export AWS_SECRET_ACCESS_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
export AWS_SESSION_TOKEN=IQoJb3JpZ2luX2VjE............
三、Python脚本集成实战：从基础命令到复杂流程
3.1 核心集成方式：subprocess模块调用
AWS CLI作为独立的命令行工具，在Python中通过subprocess模块实现调用。核心类包括：
subprocess.run()：执行命令并等待完成（推荐用于Python 3.7+）subprocess.Popen()：创建子进程并返回对象，支持异步操作
示例1：查询S3存储桶列表
import subprocess
import json
def list_s3_buckets():
"""调用aws s3 ls命令获取存储桶列表"""
# 执行命令并捕获输出
result = subprocess.run(
["aws", "s3", "ls"],
capture_output=True,
text=True
)
# 检查返回码（0表示成功）
if result.returncode != 0:
raise Exception(f"命令执行失败：{result.stderr}")
# 解析文本输出（非JSON格式，需手动处理）
buckets = []
for line in result.stdout.splitlines()[1:]: # 跳过首行标题
parts = line.split()
if len(parts) >= 3:
buckets.append({
"name": parts[2],
"creation_date": parts[0] + " " + parts[1]
})
return buckets
# 调用函数并打印结果
try:
s3_buckets = list_s3_buckets()
print("S3存储桶列表：")
for bucket in s3_buckets:
print(f"- {bucket['name']}（创建时间：{bucket['creation_date']}）")
except Exception as e:
print(f"操作失败：{str(e)}")
关键点解析：
capture_output=True：捕获标准输出（stdout）和标准错误（stderr）text=True：以字符串形式返回输出，避免字节流处理- 非JSON输出的解析逻辑：需根据命令输出格式编写定制化解析代码
示例2：通过JSON输出提升解析效率
import subprocess
import json
def list_s3_buckets_json():
"""使用--output json参数获取结构化数据"""
result = subprocess.run(
["aws", "s3api", "list-buckets", "--output", "json"],
capture_output=True,
text=True
)
if result.returncode != 0:
raise Exception(f"API调用失败：{result.stderr}")
# 直接解析JSON数据
data = json.loads(result.stdout)
return [{"name": bucket["Name"], "creation_date": bucket["CreationDate"]} for bucket in data["Buckets"]]
# 调用示例
try:
s3_buckets = list_s3_buckets_json()
print("结构化S3存储桶数据：")
for bucket in s3_buckets:
print(f"名称：{bucket['name']}，创建时间：{bucket['creation_date']}")
except Exception as e:
print(f"错误：{str(e)}")
优化点说明：
- 使用
aws s3api命令直接调用底层API，支持--output json参数 - 避免手动解析文本，提升代码健壮性与可维护性
3.2 高级操作：参数动态生成与批量处理
示例3：批量创建EC2实例
import subprocess
import random
import string
def create_ec2_instances(count=2, instance_type="t2.micro", region="us-west-2"):
"""批量创建EC2实例"""
# 生成随机标签
tag_name = "".join(random.choices(string.ascii_lowercase, k=8))
# 构建命令参数
command = [
"aws", "ec2", "run-instances",
"--region", region,
"--image-id", "ami-0c55b159cbfafe1f00", # Amazon Linux 2 AMI
"--instance-type", instance_type,
"--min-count", str(count),
"--max-count", str(count),
"--tag-specifications", f"ResourceType=instance,Tags=[{{Key=Name,Value=Python-Auto-{tag_name}}}]"
]
# 执行命令
result = subprocess.run(
command,
capture_output=True,
text=True
)
if result.returncode != 0:
raise Exception(f"实例创建失败：{result.stderr}")
# 解析实例ID
instances = json.loads(result.stdout)["Instances"]
return [instance["InstanceId"] for instance in instances]
# 调用示例
try:
instance_ids = create_ec2_instances(count=3, region="eu-north-1")
print(f"成功创建实例：{', '.join(instance_ids)}")
except Exception as e:
print(f"操作失败：{str(e)}")
技术要点：
- 动态生成标签避免名称冲突
- 使用
--tag-specifications参数进行资源标记 - 通过
--min-count与--max-count控制创建数量
示例4：批量删除未使用的EBS卷
import subprocess
import json
def delete_unattached_ebs_volumes(region="us-west-2"):
"""删除未附加的EBS卷"""
# 获取未附加的卷列表
command = [
"aws", "ec2", "describe-volumes",
"--region", region,
"--filters", "Name=status,Values=available",
"--output", "json"
]
result = subprocess.run(command, capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"查询卷列表失败：{result.stderr}")
volumes = json.loads(result.stdout)["Volumes"]
if not volumes:
print("没有未使用的EBS卷")
return
# 提取卷ID并删除
volume_ids = [volume["VolumeId"] for volume in volumes]
for volume_id in volume_ids:
delete_command = [
"aws", "ec2", "delete-volume",
"--region", region,
"--volume-id", volume_id
]
delete_result = subprocess.run(
delete_command,
capture_output=True,
text=True
)
if delete_result.returncode == 0:
print(f"成功删除卷：{volume_id}")
else:
print(f"删除卷{volume_id}失败：{delete_result.stderr}")
# 调用示例
delete_unattached_ebs_volumes(region="ap-southeast-1")
最佳实践：
- 通过
describe-volumes过滤条件精准定位目标资源 - 采用分页处理应对大规模资源列表（可通过
--page-size参数控制） - 添加干运行（Dry Run）机制：在命令中添加
--dry-run参数预验证操作
四、生产级脚本开发：错误处理与流程编排
4.1 健壮性设计：异常捕获与重试机制
import subprocess
import json
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避策略
retry=retry_if_exception_type(subprocess.CalledProcessError)
)
def retryable_aws_command(command):
"""带重试机制的AWS命令执行函数"""
result = subprocess.run(
command,
capture_output=True,
text=True,
check=True # 自动抛出CalledProcessError异常
)
return result.stdout
def example_with_retry():
"""示例：带重试的S3文件上传"""
command = [
"aws", "s3", "cp",
"local_file.txt", "s3://my-bucket/remote_path.txt",
"--region", "us-west-2"
]
try:
output = retryable_aws_command(command)
print("上传成功，输出：", output)
except subprocess.CalledProcessError as e:
print(f"最终重试失败：{e.stderr}")
except Exception as e:
print(f"其他异常：{str(e)}")
example_with_retry()
依赖库说明：
tenacity：实现灵活的重试策略，支持指数退避、自定义停止条件等check=True：启用subprocess的错误检查机制，非零返回码时自动抛异常
4.2 复杂流程编排：状态机与资源依赖
import subprocess
import json
import time
def deploy_lambda_function(function_name, zip_file_path, role_arn, region="us-east-1"):
"""Lambda函数部署全流程：创建->配置->发布"""
# 1. 创建函数
create_command = [
"aws", "lambda", "create-function",
"--region", region,
"--function-name", function_name,
"--zip-file", f"fileb://{zip_file_path}",
"--handler", "lambda_function.lambda_handler",
"--runtime", "python3.9",
"--role", role_arn,
"--output", "json"
]
create_result = subprocess.run(create_command, capture_output=True, text=True)
if create_result.returncode != 0:
raise Exception(f"创建函数失败：{create_result.stderr}")
function_arn = json.loads(create_result.stdout)["FunctionArn"]
# 2. 等待函数创建完成（轮询状态）
print("等待函数初始化...")
while True:
get_command = [
"aws", "lambda", "get-function",
"--region", region,
"--function-name", function_name,
"--output", "json"
]
get_result = subprocess.run(get_command, capture_output=True, text=True)
if get_result.returncode != 0:
time.sleep(5)
continue
status = json.loads(get_result.stdout)["Configuration"]["State"]
if status == "active":
break
time.sleep(3)
# 3. 配置环境变量（示例）
update_command = [
"aws", "lambda", "update-function-configuration",
"--region", region,
"--function-name", function_name,
"--environment", "Variables={ENV=prod,LOG_LEVEL=INFO}"
]
subprocess.run(update_command, check=True)
# 4. 发布新版本
publish_command = [
"aws", "lambda", "publish-version",
"--region", region,
"--function-name", function_name,
"--output", "json"
]
publish_result = subprocess.run(publish_command, capture_output=True, text=True)
version = json.loads(publish_result.stdout)["Version"]
print(f"成功部署Lambda函数，版本号：{version}")
# 调用示例（需提前准备好ZIP文件与IAM角色ARN）
deploy_lambda_function(
function_name="my-python-lambda",
zip_file_path="/path/to/code.zip",
role_arn="arn:aws:iam::1234567890:role/lambda-role"
)
流程关键点：
- 资源创建后的状态轮询（处理异步操作）
- 多步骤之间的依赖关系管理
- 敏感信息处理：通过参数传入而非硬编码
- 中间结果提取：从创建响应中解析Function ARN
五、典型应用场景：自动化备份与成本优化
5.1 场景一：本地数据定期同步至S3（带版本控制）
import subprocess
import datetime
def backup_to_s3(source_dir, bucket_name, region="us-west-2", keep_days=7):
"""
数据备份到S3并清理旧版本
:param source_dir: 本地源目录
:param bucket_name: S3存储桶名称
:param region: 区域
:param keep_days: 保留天数
"""
# 1. 执行同步命令（增量同步，跳过已存在文件）
sync_command = [
"aws", "s3", "sync",
source_dir, f"s3://{bucket_name}/backup/{datetime.datetime.now().strftime('%Y%m%d%H%M')}",
"--region", region,
"--delete" # 删除存储桶中不存在于本地的文件
]
subprocess.run(sync_command, check=True)
print("数据同步完成")
# 2. 清理旧版本（通过删除过期的对象版本）
list_command = [
"aws", "s3api", "list-object-versions",
"--bucket", bucket_name,
"--prefix", "backup/",
"--region", region,
"--output", "json"
]
result = subprocess.run(list_command, capture_output=True, text=True)
versions = json.loads(result.stdout).get("Versions", [])
for version in versions:
key = version["Key"]
last_modified = datetime.datetime.strptime(
version["LastModified"], "%Y-%m-%dT%H:%M:%SZ"
)
age_days = (datetime.datetime.now() - last_modified).days
if age_days > keep_days:
delete_command = [
"aws", "s3api", "delete-object",
"--bucket", bucket_name,
"--key", key,
"--version-id", version["VersionId"],
"--region", region
]
subprocess.run(delete_command, check=True)
print(f"已删除过期版本：{key} (版本号：{version['VersionId']})")
# 调用示例（需提前创建存储桶并启用版本控制）
backup_to_s3(
source_dir="/data/applogs",
bucket_name="my-backup-bucket",
keep_days=30
)
最佳实践：
- 使用时间戳作为备份路径前缀，避免版本冲突
--delete参数确保存储桶与本地目录状态一致- 通过对象版本管理实现数据回滚能力
5.2 场景二：自动停止非生产环境EC2实例
import subprocess
import json
from datetime import datetime, time
def stop_non_prod_instances(region="us-east-1", stop_time=time(22, 0)):
"""
在指定时间停止带有NonProd标签的EC2实例
:param region: 区域
:param stop_time: 停止时间（UTC时间）
"""
current_time = datetime.now().time()
if current_time > stop_time:
print("已过停止时间，跳过执行")
return
# 获取带有NonProd标签的运行中实例
command = [
"aws", "ec2", "describe-instances",
"--region", region,
"--filters", "Name=tag:Environment,Values=NonProd", "Name=instance-state-name,Values=running",
"--output", "json"
]
result = subprocess.run(command, capture_output=True, text=True)
instances = json.loads(result.stdout)
stopped_instance_ids = []
for reservation in instances["Reservations"]:
for instance in reservation["Instances"]:
instance_id = instance["InstanceId"]
stop_command = [
"aws", "ec2", "stop-instances",
"--region", region,
"--instance-ids", instance_id
]
stop_result = subprocess.run(stop_command, capture_output=True, text=True)
if stop_result.returncode == 0:
stopped_instance_ids.append(instance_id)
print(f"已停止实例：{instance_id}")
else:
print(f"停止实例{instance_id}失败：{stop_result.stderr}")
if stopped_instance_ids:
print(f"本次共停止{len(stopped_instance_ids)}个实例：{', '.join(stopped_instance_ids)}")
else:
print("没有符合条件的实例需要停止")
# 调用示例（每日22:00 UTC执行）
stop_non_prod_instances(region="ap-south-1")
扩展建议：
- 通过Cron Job或AWS CloudWatch Events定期触发脚本
- 添加通知机制（如通过SNS发送停止结果）
- 结合标签系统（Environment=NonProd）实现资源分类管理
六、相关资源
- Pypi地址：https://pypi.org/project/awscli/
（注：AWS CLI v2推荐通过官方安装程序安装，Pypi包主要用于开发环境） - Github地址：https://github.com/aws/aws-cli
（开源代码仓库，包含贡献指南与issue跟踪） - 官方文档地址：https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-welcome.html
（包含命令参考、最佳实践与故障排除指南）
通过以上内容，您已掌握AWS CLI在Python脚本中的核心应用技巧。从基础的命令行调用到复杂的自动化流程，AWS CLI凭借其与AWS生态的深度整合，成为云资源管理的必备工具。建议在实际项目中结合具体业务场景，进一步探索其与Lambda、CloudFormation等服务的协同使用，构建更智能、高效的云基础设施管理体系。
关注我，每天分享一个实用的Python自动化工具。
