Provision AWS Services from Boto3

Reading Time: 5 minutes

Hello, Readers hope all are doing well and liked my previous blogs on various DevOps tools and practices. If you have not gone through these blogs you can click here for them. Now in this blog, we will check provision AWS Services from Boto3 like ECS and ECR but first, we will understand some basic concepts or components we are going to use here.

What is Boto3

Boto3 is the Amazon Web Services (AWS) Software Development Kit (SDK) for Python which is maintained and published by Amazon Web Services. Boto3 allows Python developers to write software that makes use of services like Amazon S3 and Amazon EC2 as well as provision AWS Services from Boto3. It is complicated in many ways but it is a rich SDK for AWS.

Boto3 is written on top of botocore which is a low-level interface to the AWS API. Botocore is the basis for the AWS CLI which is also written in python. Compared to botocore, boto3 contains a lot of great objects and methods to easily work with any of AWS Services.

Core Concept of Boto3

The core concept of boto3 are:-

1. Resource: Higher level object to access AWS services.
2. Client: Low-level object to access AWS services.
3. Meta:  Object to enter into client object from resources object.
4. Session: It Stores configuration and allows you to create AWS services resources objects and client objects.
5. Collections: Tool to iterate and manipulate groups of resources.
6. Paginators: Automatic paging of response.
7. Waiters: Waiters are the way to block until a certain state has been reached.

Provision AWS ECS and ECR from Boto3

Prerequisites:

Script for provisioning ECS and ECR

We make a directory as shown in the below picture

Now we will write scripts for ECS, ECR, and IAM Role, Policies creation python script and two scripts that call these scripts in a single click for provisioning these AWS resources as well as deletion of these resources. Also a script for passing values in Environment variables in other scripts.

(Note: Assuming readers of this have a basic understanding of python)

First, we write a script for provisioning AWS ECR with the name ecr.py. 

import boto3
from .values import ECR_REPO_NAME
client = boto3.client('ecr')
def create_repo():
    try:
        response = client.create_repository(
            repositoryName=ECR_REPO_NAME,
            imageScanningConfiguration={
                'scanOnPush': True
            }
        )
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return True
        else:
            return False
    except Exception as e:
        print(str(e))
        return False

def delete_repo():
    try:
        response = client.delete_repository(
            repositoryName=ECR_REPO_NAME,
            force=True
        )
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return True
        else:
            return False
    except Exception as e:
        print(str(e))
        return False

To connect to the low-level client interface, you must use Boto3’s client(). We then pass in the name of the service you want to connect to, in this case, ecr:

import boto3
from .values import ECR_REPO_NAME
client = boto3.client('ecr')

We have defined functions for the creation of an ECR Repository and deletion of the same respectively.

Creation of ECR Repo:

def create_repo():
    try:
        response = client.create_repository(
            repositoryName=ECR_REPO_NAME,
            imageScanningConfiguration={
                'scanOnPush': True
            }
        )
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return True
        else:
            return False
    except Exception as e:
        print(str(e))
        return False

Deletion of ECR Repo

def delete_repo():
    try:
        response = client.delete_repository(
            repositoryName=ECR_REPO_NAME,
            force=True
        )
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return True
        else:
            return False
    except Exception as e:
        print(str(e))
        return False

Now, we will write a script for the rest of AWS Resources same as ECR

Provisioning AWS ECS

import boto3
from .values import (
    ECS_CLUSTER_NAME, DOCKER_IMAGE_URI, ECS_CONTAINER_NAME, 
    ECS_TASK_DEFINITION_NAME, ECS_TASK_ROLE_ARN
)

client = boto3.client('ecs')

def create_cluster():
    try:
        response = client.create_cluster(
            clusterName=ECS_CLUSTER_NAME
        )
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return True
        else:
            return False
    except Exception as e:
        print(str(e))
        return False
def create_task_definition():
    try:
        response = client.register_task_definition(
            family=ECS_TASK_DEFINITION_NAME,
            taskRoleArn=ECS_TASK_ROLE_ARN,
            executionRoleArn=ECS_TASK_ROLE_ARN,
            networkMode='awsvpc',
            containerDefinitions=[
                {
                    'name': ECS_CONTAINER_NAME,
                    'image': f'{DOCKER_IMAGE_URI}:latest',
                    'portMappings': [
                        {
                            'containerPort': 5000,
                            'protocol': 'tcp'
                        },
                    ],
                    'essential': True,
                    'logConfiguration': {
                        'logDriver': 'awslogs',
                        'options': {                            
                            'awslogs-group': f'/ecs/{ECS_TASK_DEFINITION_NAME}',
                            'awslogs-region': 'us-east-1',
                            'awslogs-stream-prefix': 'ecs'
                        }
                    }
                }
            ],
            requiresCompatibilities=[
                'FARGATE',
            ],
            cpu='256',
            memory='512',
            runtimePlatform={
                'cpuArchitecture': 'X86_64',
                'operatingSystemFamily': 'LINUX'
            }
        )
        return response['taskDefinition']
    except Exception as e:
        print(str(e))
        return False

In the ECS script, we are creating two things 

1. ECS Cluster
2. Task definition

Script for IAM role and policy creation.

import boto3
import json
from .values import ECS_TASK_POLICY_ARN, ECS_TASK_POLICY_DESC, ECS_TASK_POLICY_JSON, ECS_TASK_POLICY_NAME, ECS_TASK_ROLE_JSON, ECS_TASK_ROLE_NAME, ECS_TASK_ROLE_DESC

client = boto3.client('iam')

def create_role():
    try:
        response = client.create_role(
            RoleName=ECS_TASK_ROLE_NAME,
            AssumeRolePolicyDocument=json.dumps(ECS_TASK_ROLE_JSON),
            Description=ECS_TASK_ROLE_DESC
        )
        return response['Role']
    except Exception as e:
        print(str(e))
        return False

def create_policy():
    try:
        response = client.create_policy(
            PolicyName=ECS_TASK_POLICY_NAME,
            PolicyDocument=json.dumps(ECS_TASK_POLICY_JSON),
            Description=ECS_TASK_POLICY_DESC
        )
        return response['Policy']
    except Exception as e:
        print(str(e))
        return False

def attach_policy_to_role():
    try:
        client.attach_role_policy(
            RoleName=ECS_TASK_ROLE_NAME,
            PolicyArn=ECS_TASK_POLICY_ARN
        )
    except Exception as e:
        print(str(e))
        return False

def detach_policy_from_role():
    try:
        client.detach_role_policy(
            RoleName=ECS_TASK_ROLE_NAME,
            PolicyArn=ECS_TASK_POLICY_ARN
        )
    except Exception as e:
        print(str(e))
        return False

def delete_policy():
    try:
        client.delete_policy(
            PolicyArn=ECS_TASK_POLICY_ARN
        )
    except Exception as e:
        print(str(e))
        return False

def delete_role():
    try:
        client.delete_role(
            RoleName=ECS_TASK_ROLE_NAME
        )
    except Exception as e:
        print(str(e))
        return False

Script for Values.

 
ACCOUNT_ID='<account_id>'
ECR_REPO_NAME='bot3_ecr_demo'
ECS_CLUSTER_NAME='boto3-ecs-demo'
ECS_TASK_ROLE_NAME='ECSTaskRole'
ECS_TASK_ROLE_DESC='ECS Task Role'
ECS_TASK_POLICY_NAME='ECSTaskExceutionPolicy'
ECS_TASK_POLICY_DESC='ECS Task Execution Policy'
ECS_TASK_ROLE_JSON={
    "Version": "2012-10-17",
    "Statement": [{
            "Sid": "",
            "Effect": "Allow",
            "Principal": {
                "Service": "ecs-tasks.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}
ECS_TASK_POLICY_JSON={
    "Version": "2012-10-17",
    "Statement": [{
            "Effect": "Allow",
            "Action": [
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:GetDownloadUrlForLayer",
                "ecr:BatchGetImage",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "*"
        }
    ]
}
ECS_TASK_POLICY_ARN=f'arn:aws:iam::{ACCOUNT_ID}:policy/{ECS_TASK_POLICY_NAME}'
ECS_TASK_ROLE_ARN=f'arn:aws:iam::{ACCOUNT_ID}:role/{ECS_TASK_ROLE_NAME}'
ECS_CONTAINER_NAME='nginx'
ECS_TASK_DEFINITION_NAME='boto3_ecs_task_demo'
DOCKER_IMAGE_URI=f'{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/{ECR_REPO_NAME}'

Now We see scripts for the creation and deletion of these resources on a single hit basically these scripts call theses above scripts.  

For the creation of AWS resources

from aws_resources import ecr
from aws_resources import ecs
from aws_resources import iam

ecr.create_repo()
print('Created ECR repository')
ecs.create_cluster()
print('Created ECS Cluster')
iam.create_role()
print('Created ECS Task Role')
iam.create_policy()
print('Created ECS Task Execution Policy')
iam.attach_policy_to_role()
print('Attached policy to the role')
ecs.create_task_definition()
print('Created task definition')

Deletion of AWS resources

from aws_resources import ecr
from aws_resources import ecs
from aws_resources import iam

ecr.delete_repo()
print('Deleted ECR repo')
ecs.delete_cluster()
print('Deleted ECS Cluster')
iam.detach_policy_from_role()
print('Detached policy from the role')
iam.delete_policy()
print('Deleted policy')
iam.delete_role()
print('Deleted role')

Hands-on

Configure basic settings that the AWS CLI uses to interact with AWS as well as profile creation.

RUN: python create_aws_resources.py

Conclusion

In the blog, we saw how we can provision AWS resources through boto3 and can help the teams in provisioning infrastructure better. That’s pretty much it for this blog. If you have any feedback or queries, please do let me know in the comments. Also, if you liked the article, please give me a thumbs up and I will keep writing blogs like this for you in the future as well.

“Keep reading and Keep coding“