Innovative Software Technology

In the rapidly evolving world of DevOps, automation is paramount. While some might assume DevOps engineers shy away from coding, the reality is quite the opposite. Crafting scripts and tools to streamline repetitive tasks, manage infrastructure as code, and orchestrate complex cloud workflows is a core part of a DevOps professional’s daily routine. This article embarks on a practical journey within our ‘Python for DevOps’ series, guiding you through the creation of a custom Command Line Interface (CLI) tool. This powerful utility, built with Python3 and the AWS SDK Boto3, will enable you to effortlessly manage your Amazon EC2 instances, S3 buckets, and Lambda functions directly from your terminal. By the end of this project, you’ll not only enhance your Python proficiency and gain hands-on experience with Boto3, but also learn to automate essential AWS operations like a seasoned expert.

📽️ Youtube Demonstration

🧰 Prerequisites

Before diving into the development of our custom AWS CLI, ensure your environment is adequately prepared. You will need:

  • 🪄 AWS CLI Installed and Configured: A functional AWS CLI setup is essential. It must be configured with an IAM user possessing full access permissions for EC2, Lambda, IAM, and S3 services. If you need assistance with this setup, refer to ‘Step 1’ of this guide: Learn how to deploy a Three-Tier Application on AWS EKS using Terraform.
  • 🐍 Python 3: Confirm that Python 3 is installed on your system. You can verify your version by running:

    python3 --version

Once these prerequisites are met, we can proceed to structure our project and begin coding.

🧠 Writing the Core Python Scripts — EC2, S3 & Lambda Automation

The heart of our AWS automation CLI lies in several modular Python scripts, each dedicated to managing a specific AWS service. The complete source code for this project is hosted on my GitHub repository: Pravesh-Sudha/python-for-devops. Navigate to the automate-ec2-s3-and-lambda directory to explore the following foundational files:

  • ec2_manager.py
  • s3_manager.py
  • lambda_manager.py

These scripts leverage Boto3, the official AWS SDK for Python, to interact programmatically with AWS services.

1. EC2 Manager — ec2_manager.py

This script provides functionalities to lifecycle manage EC2 instances.

import boto3
from botocore.exceptions import ClientError

class EC2Manager:
    def __init__(self, region_name="us-east-1"):
        self.ec2 = boto3.client("ec2", region_name=region_name)

    def create_instance(self, image_id="ami-0360c520857e3138f", instance_type="t3.micro", key_name=None):
        try:
            print("Creating a Ec2 Instance....")
            params = {
                "ImageId": image_id,
                "InstanceType": instance_type,
                "MinCount": 1,
                "MaxCount": 1
            }
            if key_name:
                params["KeyName"] = key_name

            response = self.ec2.run_instances(**params)
            instance_id = response["Instances"][0]["InstanceId"]
            print(f"EC2 instance Created successfully with id: {instance_id}")
            return instance_id
        except ClientError as e:
            print(f"Ec2 creation failed. Error occured: {e}")
            return None

    def list_instances(self):
        try:
            response = self.ec2.describe_instances()
            instances = []
            for reservation in response["Reservations"]:
                for instance in reservation["Instances"]:
                    instance_info = {
                        "InstanceId": instance["InstanceId"],
                        "State": instance["State"]["Name"],
                        "Type": instance["InstanceType"],
                        "PublicIP": instance.get("PublicIpAddress")
                    }
                    instances.append(instance_info)
            return instances
        except ClientError as e:
            print(f"Failed to get list of Ec2 instances. Error Occured: {e}")
            return []

    def terminate_instance(self, instance_id):
        try:
            self.ec2.terminate_instances(InstanceIds=[instance_id])
            print(f"Instance with id {instance_id} terminated successfully")
        except ClientError as e:
            print(f"Failed to terminate instance {instance_id}. Error Occured: {e}")

Functionality Overview:

  • Initializes a Boto3 EC2 client.
  • create_instance(): Provisions a new EC2 instance, allowing customization of AMI ID, instance type, and key pair.
  • list_instances(): Retrieves and displays details of all active EC2 instances, including their ID, state, type, and public IP.
  • terminate_instance(): Safely terminates a specified EC2 instance by its ID.
  • Robust error handling with try-except blocks ensures graceful operation even when AWS API calls encounter issues.

2. S3 Manager — s3_manager.py

This module simplifies the management of S3 buckets.

import boto3
from botocore.exceptions import ClientError

class S3Manager:
    def __init__(self, region_name="us-east-1"):
        self.s3 = boto3.client("s3", region_name=region_name)

    def create_bucket(self, bucket_name):
        try:
            print(f"Creating a new Bucket name: {bucket_name}")
            self.s3.create_bucket(Bucket=bucket_name)
            print(f"Bucket Created Successfully.")
        except ClientError as e:
            print(f"Failed to create Bucket. Error: {e}")

    def list_buckets(self):
        try:
            response = self.s3.list_buckets()
            buckets = [b["Name"] for b in response["Buckets"]]
            return buckets
        except ClientError as e:
            print(f"Failed to list the bucket. Error: {e}")
            return []

    def delete_bucket(self, bucket_name):
        try:
            print(f"Deleting Bucket: {bucket_name}")
            self.s3.delete_bucket(Bucket=bucket_name)
            print(f"Successfully deleted the bucket: {bucket_name}")
        except ClientError as e:
            print(f"Failed to delete the Bucket. Error: {e}")

Functionality Overview:

  • Sets up a Boto3 S3 client.
  • create_bucket(): Creates a new S3 bucket with the specified name.
  • list_buckets(): Fetches and lists the names of all S3 buckets in your account.
  • delete_bucket(): Removes an S3 bucket by its name.
  • Error handling is implemented to manage potential issues during bucket operations.

3. Lambda Manager — lambda_manager.py

The Lambda manager handles the creation, listing, and deletion of AWS Lambda functions, including the necessary IAM role management.

import boto3
from botocore.exceptions import ClientError
import zipfile
import os
import time, json

class LambdaManager:
    def __init__(self, region_name="us-east-1"):
        self.lambda_client = boto3.client("lambda", region_name = region_name)
        self.iam_client = boto3.client("iam", region_name = region_name)

    def _create_deployment_package(self, code_path, zip_name="lambda_function.zip"):
        with zipfile.ZipFile(zip_name, "w") as zf:
            zf.write(code_path, os.path.basename(code_path))
            print(f"Deployment package built Successfully")
        return zip_name

    def _get_or_create_role(self, role_name="LambdaBasicExecutionRole"):
        assume_role_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"Service": "lambda.amazonaws.com"},
                    "Action": "sts:AssumeRole"
                }
            ]
        }
        try:
            role = self.iam_client.get_role(RoleName = role_name)
            print(f"Getting the IAM role for Lambda....")
            return role["Role"]["Arn"]
        except self.iam_client.exceptions.NoSuchEntityException as e:
            print(f"Creating the IAM role : {role_name}....")
            role = self.iam_client.create_role(
                RoleName = role_name,
                AssumeRolePolicyDocument = json.dumps(assume_role_policy) 
            )
            self.iam_client.attach_role_policy(
                RoleName = role_name,
                PolicyArn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
            )
            print(f"Waiting for the IAM role Propagation...")
            time.sleep(10)
            return role["Role"]["Arn"]

    def create_lambda(self, function_name, code_path, handler_name="lambda_function.lambda_handler", runtime = "python3.9"):        
        try:
            zip_file = self._create_deployment_package(code_path)
            role_arn = self._get_or_create_role()

            with open(zip_file, "rb") as f:
                zip_bytes = f.read()

            response = self.lambda_client.create_function(
                FunctionName = function_name,
                Code = {"ZipFile": zip_bytes},
                Runtime = runtime,
                Role = role_arn,
                Handler = handler_name,
                Timeout = 15,
                MemorySize = 128, 
            )
            print(f"Lambda function {function_name} created Successfully.")
            return response["FunctionArn"]
        except ClientError as e:
            print(f"Failed to create the lambda function. Error: {e}")

    def list_lambdas(self):
        try:
            response = self.lambda_client.list_functions()
            lambdas = response.get("Functions",[])
            return [
                {
                "Name": fn["FunctionName"],
                "Runtime": fn["Runtime"],
                "Arn": fn["FunctionArn"]
                }
                for fn in lambdas
            ]
        except ClientError as e:
            print(f"Failed to list Lambda Functions. Error: {e}")
            return []

    def delete_lambda(self, function_name):
        try:
            self.lambda_client.delete_function(FunctionName = function_name)
            print(f"Lambda function: {function_name} deleted Successfully.")
        except ClientError as e:
            print(f"Failed to delete Lambda Function. Error: {e}")

Functionality Overview:

  • Initializes Boto3 Lambda and IAM clients.
  • _create_deployment_package(): Compresses the Python function code into a .zip file, a requirement for Lambda deployments.
  • _get_or_create_role(): Manages the IAM role necessary for Lambda function execution. It checks for an existing role or creates a new one with appropriate permissions, ensuring proper propagation before proceeding.
  • create_lambda(): Deploys a new Lambda function using the zipped code and the managed IAM role.
  • list_lambdas(): Retrieves details of all deployed Lambda functions.
  • delete_lambda(): Removes a specified Lambda function.
  • Includes a sample lambda_function.py for testing purposes: 
    import json

def lambda_handler(event, context):
    print("Hello from Pravesh - Your AWS Community Builder!") 
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Pravesh - Your AWS Community Builder!')
    }

4. Orchestrating with main.py (CLI Interface)

To transform these scripts into a user-friendly command-line tool, we integrate them using Python’s argparse module. This allows for intuitive command execution patterns, similar to the native AWS CLI.

For instance:

python main.py ec2 create
python main.py s3 list
python main.py lambda delete <function-name>

Here’s the comprehensive main.py script:

import argparse
from ec2_manager import EC2Manager
from s3_manager import S3Manager
from lambda_manager import LambdaManager

def main():
    parser = argparse.ArgumentParser(
        description="AWS Automation CLI using Boto3 (EC2 & S3)"
    )

    subparsers = parser.add_subparsers(dest="service", help="Choose AWS service")

    # EC2 Commands
    ec2_parser = subparsers.add_parser("ec2", help="Manage EC2 instances")
    ec2_subparsers = ec2_parser.add_subparsers(dest="action", help="EC2 actions")

    ec2_create = ec2_subparsers.add_parser("create", help="Create EC2 instance")
    ec2_create.add_argument("--image-id", default="ami-0c94855ba95c71c99", help="AMI ID")
    ec2_create.add_argument("--instance-type", default="t2.micro", help="Instance type")
    ec2_create.add_argument("--key-name", help="Optional key pair name")

    ec2_subparsers.add_parser("list", help="List EC2 instances")
    ec2_terminate = ec2_subparsers.add_parser("terminate", help="Terminate EC2 instance")
    ec2_terminate.add_argument("instance_id", help="Instance ID to terminate")

    # S3 Commands
    s3_parser = subparsers.add_parser("s3", help="Manage S3 buckets")
    s3_subparsers = s3_parser.add_subparsers(dest="action", help="S3 actions")
    s3_subparsers.add_parser("list", help="List S3 buckets")
    s3_create = s3_subparsers.add_parser("create", help="Create S3 bucket")
    s3_create.add_argument("bucket_name", help="Bucket name to create")
    s3_delete = s3_subparsers.add_parser("delete", help="Delete S3 bucket")
    s3_delete.add_argument("bucket_name", help="Bucket name to delete")

    # Lambda Commands
    lambda_parser = subparsers.add_parser("lambda", help="Manage AWS Lambda functions")
    lambda_subparsers = lambda_parser.add_subparsers(dest="action", help="Lambda actions")
    lambda_create = lambda_subparsers.add_parser("create", help="Create Lambda function")
    lambda_create.add_argument("function_name", help="Lambda function name")
    lambda_create.add_argument("code_path", help="Path to Python file (e.g. lambda_function.py)")
    lambda_create.add_argument("--handler", default="lambda_function.lambda_handler", help="Handler name")
    lambda_create.add_argument("--runtime", default="python3.9", help="Runtime version")
    lambda_subparsers.add_parser("list", help="List Lambda functions")
    lambda_delete = lambda_subparsers.add_parser("delete", help="Delete Lambda function")
    lambda_delete.add_argument("function_name", help="Lambda function name to delete")

    args = parser.parse_args()

    # Service Handling
    if args.service == "ec2":
        ec2 = EC2Manager()
        if args.action == "create":
            ec2.create_instance(image_id=args.image_id, instance_type=args.instance_type, key_name=args.key_name)
        elif args.action == "list":
            instances = ec2.list_instances()
            if not instances:
                print("No EC2 instances found.")
            else:
                print("
EC2 Instances:")
                for i in instances:
                    print(f"{i['InstanceId']} | {i['State']} | {i['Type']} | {i.get('PublicIP', 'N/A')}")
        elif args.action == "terminate":
            ec2.terminate_instance(args.instance_id)

    elif args.service == "s3":
        s3 = S3Manager()
        if args.action == "create":
            s3.create_bucket(args.bucket_name)
        elif args.action == "list":
            buckets = s3.list_buckets()
            if not buckets:
                print("No buckets found.")
            else:
                print("
S3 Buckets:")
                for b in buckets:
                    print(f"- {b}")
        elif args.action == "delete":
            s3.delete_bucket(args.bucket_name)

    elif args.service == "lambda":
        lm = LambdaManager()
        if args.action == "create":
            lm.create_lambda(args.function_name, args.code_path, handler_name=args.handler, runtime=args.runtime)
        elif args.action == "list":
            functions = lm.list_lambdas()
            if not functions:
                print("No Lambda functions found.")
            else:
                print("
Lambda Functions:")
                for f in functions:
                    print(f"{f['Name']} | {f['Runtime']} | {f['Arn']}")
        elif args.action == "delete":
            lm.delete_lambda(args.function_name)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()

This main.py script serves as the central dispatcher, parsing command-line arguments and invoking the appropriate methods from EC2Manager, S3Manager, and LambdaManager classes to execute the desired AWS operations.

🧠 Pro Tip — Write the Code Yourself

Empower Yourself: Code, Don’t Just Copy

A fundamental principle for true mastery in development is active engagement. Instead of merely copying and pasting the provided code, I strongly encourage you to type out each line of ec2_manager.py within your own development environment. This deliberate process helps you internalize:

  • The strategic use of try/except blocks for robust error handling.
  • The mechanics of initializing boto3 clients.
  • The structure and invocation of AWS API calls.

Once comfortable, extend this practice to the S3 and Lambda manager files. Refer to the repository for guidance when you encounter challenges, but prioritize independent coding. This hands-on approach cultivates muscle memory and builds genuine confidence – invaluable assets in an era of AI-driven code generation. Understanding the ‘how’ and ‘why’ behind the code empowers you far beyond what any automated tool can offer.

🛠️ Setting Up the Virtual Environment

To ensure a clean and isolated development environment for our CLI tool, set up a Python virtual environment and install the necessary dependencies:

python3 -m venv venv
source venv/bin/activate
pip3 install boto3 botocore

🚀 Running the CLI Commands

With your environment configured, you can now unleash the power of your custom AWS automation CLI. Here are examples of how to interact with EC2, S3, and Lambda services:

EC2

python main.py ec2 create --image-id ami-0c94855ba95c71c99 --instance-type t2.micro --key-name your-key-pair
python main.py ec2 list
python main.py ec2 terminate <instance-id>

S3

python main.py s3 create <unique-bucket-name>
python main.py s3 list
python main.py s3 delete <unique-bucket-name>

Lambda

python main.py lambda create <function-name> lambda_function.py
python main.py lambda list
python main.py lambda delete <function-name>

Testing Your Lambda Function:

Before proceeding with deletion, it’s a good practice to test your newly created Lambda function directly from the AWS Lambda Dashboard. Navigate to your function, configure a test event (you can use the default ‘hello-world’ template), and invoke it.

You should observe the output from your lambda_function.py (e.g., “Hello from Pravesh – Your AWS Community Builder!”) confirming successful deployment and execution.

🏁 Conclusion

Congratulations! You’ve successfully developed a fully functional Python CLI tool capable of orchestrating your AWS EC2 instances, S3 buckets, and Lambda functions directly from your command line. This project, seemingly straightforward, offers profound insights into how DevOps engineers leverage programming to simplify and automate day-to-day cloud operations.

This is merely the foundation; the possibilities for expanding this CLI are vast. Consider integrating:

  • Support for additional AWS services like RDS, DynamoDB, or CloudWatch for comprehensive monitoring and management.
  • Advanced features such as enhanced error reporting, detailed logging, or interactive prompts.
  • Granular, role-based access control through more intricate IAM policies.

The entire project’s source code is available on GitHub. Your forks and stars are incredibly motivating and help inspire more Python + DevOps content. If you’ve crafted your own version or extended its capabilities, please share it with me or tag me on social media:

Keep automating, keep building, and continue pushing the boundaries of what’s possible with code! 💪⚡

Leave a Reply

Your email address will not be published. Required fields are marked *

Fill out this field
Fill out this field
Please enter a valid email address.
You need to agree with the terms to proceed