AWS CloudFormation 系列--(3)Python操作CloudFormation

AWS CloudFormation 系列--(3)Python操作CloudFormation
B站视频

前面两篇已经对CloudFormation做了快速入门和进阶的学习,但是在创建堆栈的时候,还是在AWS控制台界面导入文本文件。这个文章的目的是介绍如何使用Python来操作CloudFormation,进一步的实现自动化操作。

一、AWS 凭证(Credentials)

1.1、凭证介绍

客户端必须拥有凭证[参考链接1]才能对AWS API进行操作,可以通过多种方式来配置凭证。配置凭证的最佳实践是,如果你在EC2上运行代码,建议使用IAM角色。如果你是用AWS开发工具包(例如Python),建议使用共享凭证文件。后面代码都是使用的共享凭证文件。

Boto3 将在搜索凭证时查找多个位置。Boto3 查找凭据的机制是搜索可能位置的列表,并在找到凭据后立即停止。Boto3 搜索凭据的顺序参考:

1.在boto.client()方法中将凭证作为参数传递
2.创建Session对象时将凭证作为参数传递
3.环境变量
4.共享凭证文件 ( ~/.aws/credentials )
5.AWS 配置文件 ( ~/.aws/config )
6.担任角色提供者
7.Boto2 配置文件(/etc/boto.cfg和~/.boto)
8.配置了 IAM 角色的 Amazon EC2 实例上的实例元数据服务。

1.2、AWS CLI配置凭证

安装AWS CLI[参考链接2]客户端之后,可以利用aws configure快速配置凭证。需要提前从AWS控制台创建对应账号的AK(Access Key)和SK(Secret Key)。

$ aws configure
AWS Access Key ID [None]: AKIAIOSFODNN7EXAMPLE
AWS Secret Access Key [None]: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
Default region name [None]: cn-northwest-1
Default output format [None]: 

直接使用aws configure产生的是default配置文件,在没有明确指定配置文件时,就会使用default配置文件。你还可以使用aws configure --profile cn-prod来创建名称为cn-prod的配置文件。假设default配置的是测试账号信息,cn-prod配置的是生产账号信息,在不明确指定配置文件的情况下,代码会使用default配置,也就是测试环境的凭证。

配置凭证时,指定名称为cn-prod

$ aws configure --profile cn-prod
AWS Access Key ID [None]: AKIAI44QH8DHBEXAMPLE
AWS Secret Access Key [None]: je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY
Default region name [None]: cn-northwest-1
Default output format [None]: 

AWS会将配置文件放在.aws文件夹中,这个文件夹默认在用户主目录下,例如Windows的路径C:\Users\crosswalk\.aws,Linux root用户路径/root/.aws。文件夹中会产生一个凭证配置文件credentials和一个Config配置文件config

凭证配置文件credentials中具体的内容:

[default]
aws_access_key_id=AKIAIOSFODNN7EXAMPLE
aws_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

[cn-prod]
aws_access_key_id=AKIAI44QH8DHBEXAMPLE
aws_secret_access_key=je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY

Config配置文件config中的具体内容:

[default]
region=cn-northwest-1

[profile cn-prod]
region=cn-northwest-1

使用 aws configure list-profiles 命令列出所有配置文件名称。

C:\Users\crosswalk>aws configure list-profiles
default
cn-prod

1.3、AWS CLI凭证测试

使用默认凭证,查看测试账号下的EC2

C:\Users\crosswalk>aws ec2 describe-instances
{
    "Reservations": [
        {
            "Groups": [],
            "Instances": [
                {
                    "AmiLaunchIndex": 0,
                    "ImageId": "ami-007315f06e322f1ab",
                    "InstanceId": "i-08fc1f70383191bb4",
                    "InstanceType": "t2.2xlarge"

指定cn-prod凭证,查看生产账号下的EC2

C:\Users\crosswalk>aws ec2 describe-instances --profile cn-prod
{
    "Reservations": [
        {
            "Groups": [],
            "Instances": [
                {
                    "AmiLaunchIndex": 0,
                    "ImageId": "ami-7c9f8b1e",
                    "InstanceId": "i-099540680995ce588",
                    "InstanceType": "t2.medium",

1.4、boto3 切换凭证测试

你可以在创建Session[参考链接3]时通过profile_name参数指定凭证配置文件名称。

profile_name = 'my-global'
my_region = 'cn-northwest-1'

session = boto3.Session(profile_name=profile_name, region_name=my_region)
dev_s3_client = session.client('s3')

在创建脚本时,开头定义使用的凭证名称。

profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_create_stack(stackname, file_path):
    """创建堆栈"""
    import boto3
    session = boto3.Session(profile_name=profile_name, region_name=my_region)
    client = session.client('cloudformation')
    with open(f"{file_path}", 'r', encoding='UTF-8') as fd:    # 读取CloudFormation YAML文件内容
        templatebody = fd.read()
    try:
        client.create_stack(StackName=f"{stackname}", Capabilities=['CAPABILITY_IAM'], TemplateBody=templatebody)
        print(f'["{stackname}"堆栈创建成功]')
    except Exception as e:
        print(f'["{stackname}"堆栈创建失败]')
        print(e)


if __name__ == "__main__":
    cft_create_stack('vpc-stack-test', 'vpc-test.yaml')

二、Python Boto3操作CloudFormation代码参考

利用Python3的Boto3模块,就可以完成AWS上的所有API操作,具体可以参考boto3 CloudFormaionl[参考链接4],查看API的详细参数。

写一个用于测试脚本的CloudFormation模板vpc-test.yaml,创建一个VPC并且输出VPC的ID信息。

Resources:
  MyTestVpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
      EnableDnsSupport: 'true'
      EnableDnsHostnames: 'true'
      Tags:
       - Key: Name
         Value: MyTestVpc

Outputs:
  MyTestVpc:
    Description: MyTestVpc
    Value: !Ref MyTestVpc
    Export:
      Name: MyTestVpc-ID

2.1、创建堆栈

脚本需要指定堆栈的名称和模板的路径。

profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_create_stack(stackname, file_path):
    """创建堆栈"""
    import boto3
    session = boto3.Session(profile_name=profile_name, region_name=my_region)
    client = session.client('cloudformation')
    with open(f"{file_path}", 'r', encoding='UTF-8') as fd:    # 读取CloudFormation YAML文件内容
        templatebody = fd.read()
    try:
        client.create_stack(StackName=f"{stackname}", Capabilities=['CAPABILITY_IAM'], TemplateBody=templatebody)
        print(f'["{stackname}"堆栈创建成功]')
    except Exception as e:
        print(f'["{stackname}"堆栈创建失败]')
        print(e)


if __name__ == "__main__":
    cft_create_stack('vpc-stack-test', 'vpc-test.yaml')

代码输出:

["vpc-stack-test"堆栈创建成功]

2.2、获取堆栈输出

profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_stack_output(stackname):    # 当堆栈完成后获取输出
    """获取堆栈输出内容"""
    import time
    import boto3
    while True:
        nowtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        session = boto3.Session(profile_name=profile_name, region_name=my_region)
        client = session.client('cloudformation')
        response = client.describe_stacks(StackName=stackname)
        if response['Stacks'][0]['StackStatus'] == 'CREATE_COMPLETE':
            response = client.describe_stacks(StackName=stackname)
            outputs = response["Stacks"][0]["Outputs"]
            print(f'[{nowtime}] Stack Create Complete!')
            print(outputs)
            return outputs
        else:
            time.sleep(1)
            print(f'[{nowtime}] Wait Stack Complete...')
            
if __name__ == "__main__":
    # cft_create_stack('vpc-stack-test', 'vpc-test.yaml')
    cfn_stack_output('vpc-stack-test')

代码输出:

[2022-04-02 11:51:26] Stack Create Complete!
[{'OutputKey': 'MyTestVpc', 'OutputValue': 'vpc-0e764b67068222644', 'Description': 'MyTestVpc', 'ExportName': 'MyTestVpc-ID'}]

2.3、删除堆栈

profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_delete_stack(stackname):
    """删除堆栈"""
    import boto3
    session = boto3.Session(profile_name=profile_name, region_name=my_region)
    client = session.client('cloudformation')
    try:
        rsp = client.delete_stack(StackName=stackname)
        print(f'["{stackname}"堆栈删除成功]')
    except Exception as e:
        print(f'["{stackname}"堆栈删除失败]')
        print(e)
        
if __name__ == "__main__":
    # cft_create_stack('vpc-stack-test', 'vpc-test.yaml')
    # cfn_stack_output('vpc-stack-test')
    cfn_delete_stack('vpc-stack-test')

代码输出:

["vpc-stack-test"堆栈删除成功]

三、参考文档