Automate AWS Cloud Infrastructure Using Python and Boto3

Cloud providers offer powerful APIs for managing infrastructure, but clicking through web consoles wastes time. Python’s Boto3 library gives you programmatic control over Amazon Web Services, letting you launch servers, store files, configure permissions, and set up alerts through clean Python code.

This tutorial walks through practical examples. You’ll connect to AWS, manage EC2 instances, handle S3 buckets, work with IAM, set up CloudWatch monitoring, and build a complete automation script. By the end, you’ll have working code for common infrastructure tasks.

Prerequisites

Before starting, you need an AWS account and Python installed. Boto3 works with Python 3.7 or later.

Install Boto3 using pip:

pip install boto3

Next, configure AWS credentials. The AWS CLI provides the easiest setup method:

pip install awscli
aws configure

Enter your Access Key ID, Secret Access Key, default region (like us-east-1), and output format (use json). These credentials get stored in ~/.aws/credentials.

Never hardcode credentials in your scripts. Boto3 automatically reads from several locations:

1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
2. The shared credentials file (~/.aws/credentials)
3. IAM roles (when running on EC2 instances)

For development, use the credentials file. For production, use IAM roles attached to your EC2 instances or Lambda functions.

Step 1: Connecting to AWS and Managing EC2 Instances

EC2 instances are virtual servers. Boto3 provides two interfaces for AWS services: clients (low-level) and resources (high-level). Resources offer a more Pythonic API.

Here’s how to connect and list your EC2 instances:

import boto3

# Create EC2 resource
ec2 = boto3.resource('ec2', region_name='us-east-1')

# List all instances
print("Current EC2 Instances:")
for instance in ec2.instances.all():
    print(f"ID: {instance.id}")
    print(f"Type: {instance.instance_type}")
    print(f"State: {instance.state['Name']}")
    print(f"Public IP: {instance.public_ip_address}")
    print("---")

To launch a new instance, specify an AMI ID and instance type:

# Launch a new Ubuntu instance
instances = ec2.create_instances(
    ImageId='ami-0c55b159cbfafe1f0',  # Ubuntu 20.04 AMI
    MinCount=1,
    MaxCount=1,
    InstanceType='t2.micro',
    KeyName='my-key-pair',
    SecurityGroupIds=['sg-0123456789abcdef0'],
    TagSpecifications=[
        {
            'ResourceType': 'instance',
            'Tags': [
                {'Key': 'Name', 'Value': 'Boto3-Test-Server'},
                {'Key': 'Environment', 'Value': 'Development'}
            ]
        }
    ]
)

new_instance = instances[0]
print(f"Launched instance: {new_instance.id}")

# Wait until the instance is running
new_instance.wait_until_running()
new_instance.reload()
print(f"Instance is now running at {new_instance.public_ip_address}")

To stop or terminate instances:

# Stop an instance (can be restarted)
instance = ec2.Instance('i-0123456789abcdef0')
instance.stop()
print("Stopping instance...")

# Terminate an instance (permanent)
instance.terminate()
print("Terminating instance...")

You can filter instances by tags or state:

# Find all running instances tagged 'Environment: Production'
filters = [
    {'Name': 'tag:Environment', 'Values': ['Production']},
    {'Name': 'instance-state-name', 'Values': ['running']}
]

prod_instances = ec2.instances.filter(Filters=filters)
for instance in prod_instances:
    print(f"Production instance: {instance.id}")

Step 2: Automating S3 Bucket Operations

S3 stores files as objects in buckets. Boto3 makes it simple to upload, download, and manage these files.

Create a bucket and upload a file:

import boto3

s3 = boto3.resource('s3')

# Create a new bucket
bucket_name = 'my-automation-bucket-2026'
s3.create_bucket(Bucket=bucket_name)
print(f"Created bucket: {bucket_name}")

# Upload a file
s3.Bucket(bucket_name).upload_file(
    Filename='local-file.txt',
    Key='remote-file.txt'
)
print("File uploaded successfully")

Bucket names must be globally unique across all AWS accounts. If the name exists, you’ll get an error.

To list and download files:

# List all objects in a bucket
bucket = s3.Bucket(bucket_name)
print("Files in bucket:")
for obj in bucket.objects.all():
    print(f"  {obj.key} ({obj.size} bytes)")

# Download a file
bucket.download_file(
    Key='remote-file.txt',
    Filename='downloaded-file.txt'
)
print("File downloaded")

For large files, use multipart uploads. Boto3 handles this automatically when you use upload_file():

# Upload a large file (multipart upload happens automatically)
s3.Bucket(bucket_name).upload_file(
    Filename='large-video.mp4',
    Key='videos/large-video.mp4',
    ExtraArgs={'StorageClass': 'STANDARD_IA'}
)

Set up lifecycle rules to move old files to cheaper storage:

# Configure lifecycle to move files to Glacier after 30 days
s3_client = boto3.client('s3')

lifecycle_policy = {
    'Rules': [
        {
            'Id': 'Move to Glacier',
            'Status': 'Enabled',
            'Prefix': 'archives/',
            'Transitions': [
                {
                    'Days': 30,
                    'StorageClass': 'GLACIER'
                }
            ]
        }
    ]
}

s3_client.put_bucket_lifecycle_configuration(
    Bucket=bucket_name,
    LifecycleConfiguration=lifecycle_policy
)
print("Lifecycle policy applied")

Delete objects and buckets:

# Delete an object
s3.Object(bucket_name, 'remote-file.txt').delete()

# Delete all objects in a bucket
bucket = s3.Bucket(bucket_name)
bucket.objects.all().delete()

# Delete the bucket itself
bucket.delete()
print("Bucket deleted")

Step 3: Working with IAM Programmatically

IAM controls who can access your AWS resources. You can create users, groups, and policies through Boto3.

Create a new IAM user:

import boto3
import json

iam = boto3.client('iam')

# Create a user
username = 'automation-user'
iam.create_user(UserName=username)
print(f"Created user: {username}")

# Create access keys
response = iam.create_access_key(UserName=username)
access_key = response['AccessKey']

print(f"Access Key ID: {access_key['AccessKeyId']}")
print(f"Secret Access Key: {access_key['SecretAccessKey']}")
print("Store these credentials securely!")

Store generated credentials in a secure location like AWS Secrets Manager or a password manager. Never commit them to version control.

Attach a policy to grant permissions:

# Define a policy allowing S3 read-only access
policy_document = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::my-automation-bucket-2026",
                "arn:aws:s3:::my-automation-bucket-2026/*"
            ]
        }
    ]
}

# Create the policy
policy_name = 'S3ReadOnlyPolicy'
policy_response = iam.create_policy(
    PolicyName=policy_name,
    PolicyDocument=json.dumps(policy_document)
)

policy_arn = policy_response['Policy']['Arn']

# Attach policy to user
iam.attach_user_policy(
    UserName=username,
    PolicyArn=policy_arn
)
print(f"Attached policy {policy_name} to {username}")

List users and their attached policies:

# List all IAM users
response = iam.list_users()
for user in response['Users']:
    print(f"User: {user['UserName']}")
    
    # List attached policies
    policies = iam.list_attached_user_policies(UserName=user['UserName'])
    for policy in policies['AttachedPolicies']:
        print(f"  Policy: {policy['PolicyName']}")

Create groups for easier management:

# Create a group
group_name = 'developers'
iam.create_group(GroupName=group_name)

# Add user to group
iam.add_user_to_group(
    GroupName=group_name,
    UserName=username
)

# Attach policy to group instead of individual users
iam.attach_group_policy(
    GroupName=group_name,
    PolicyArn=policy_arn
)
print(f"User {username} added to {group_name} group")

Step 4: Monitoring with CloudWatch

CloudWatch collects metrics and logs from AWS services. You can query metrics, create alarms, and send notifications.

Get CPU usage for an EC2 instance:

import boto3
from datetime import datetime, timedelta

cloudwatch = boto3.client('cloudwatch')

# Get average CPU for the last hour
response = cloudwatch.get_metric_statistics(
    Namespace='AWS/EC2',
    MetricName='CPUUtilization',
    Dimensions=[
        {
            'Name': 'InstanceId',
            'Value': 'i-0123456789abcdef0'
        }
    ],
    StartTime=datetime.utcnow() - timedelta(hours=1),
    EndTime=datetime.utcnow(),
    Period=300,  # 5 minutes
    Statistics=['Average']
)

print("CPU Usage (last hour):")
for datapoint in response['Datapoints']:
    print(f"{datapoint['Timestamp']}: {datapoint['Average']:.2f}%")

Create an alarm that triggers when CPU exceeds 80%:

sns = boto3.client('sns')

# First, create an SNS topic for notifications
topic_response = sns.create_topic(Name='high-cpu-alerts')
topic_arn = topic_response['TopicArn']

# Subscribe your email to the topic
sns.subscribe(
    TopicArn=topic_arn,
    Protocol='email',
    Endpoint='you@example.com'
)
print("Check your email and confirm the subscription")

# Create the CloudWatch alarm
cloudwatch.put_metric_alarm(
    AlarmName='HighCPUUsage',
    ComparisonOperator='GreaterThanThreshold',
    EvaluationPeriods=2,
    MetricName='CPUUtilization',
    Namespace='AWS/EC2',
    Period=300,
    Statistic='Average',
    Threshold=80.0,
    ActionsEnabled=True,
    AlarmActions=[topic_arn],
    AlarmDescription='Alert when CPU exceeds 80%',
    Dimensions=[
        {
            'Name': 'InstanceId',
            'Value': 'i-0123456789abcdef0'
        }
    ]
)
print("Alarm created successfully")

Push custom metrics to CloudWatch:

# Send a custom metric
cloudwatch.put_metric_data(
    Namespace='MyApplication',
    MetricData=[
        {
            'MetricName': 'ProcessedRecords',
            'Value': 150,
            'Unit': 'Count',
            'Timestamp': datetime.utcnow()
        }
    ]
)
print("Custom metric sent")

List existing alarms:

# Get all alarms
response = cloudwatch.describe_alarms()

for alarm in response['MetricAlarms']:
    print(f"Alarm: {alarm['AlarmName']}")
    print(f"  State: {alarm['StateValue']}")
    print(f"  Metric: {alarm['MetricName']}")
    print(f"  Threshold: {alarm['Threshold']}")
    print("---")

Step 5: Building a Complete Automation Script

Here’s a full script that combines what we’ve covered. This script launches an EC2 instance, uploads a file to S3, creates an IAM user with appropriate permissions, and sets up monitoring.

#!/usr/bin/env python3
import boto3
import json
import time
from datetime import datetime

class AWSAutomation:
    def __init__(self, region='us-east-1'):
        self.region = region
        self.ec2 = boto3.resource('ec2', region_name=region)
        self.s3 = boto3.resource('s3')
        self.iam = boto3.client('iam')
        self.cloudwatch = boto3.client('cloudwatch', region_name=region)
        self.sns = boto3.client('sns', region_name=region)
        
    def launch_instance(self, name, instance_type='t2.micro'):
        """Launch an EC2 instance with monitoring enabled"""
        print(f"Launching {instance_type} instance...")
        
        instances = self.ec2.create_instances(
            ImageId='ami-0c55b159cbfafe1f0',
            MinCount=1,
            MaxCount=1,
            InstanceType=instance_type,
            Monitoring={'Enabled': True},
            TagSpecifications=[
                {
                    'ResourceType': 'instance',
                    'Tags': [{'Key': 'Name', 'Value': name}]
                }
            ]
        )
        
        instance = instances[0]
        print(f"Instance {instance.id} launching...")
        instance.wait_until_running()
        instance.reload()
        
        print(f"Instance running at {instance.public_ip_address}")
        return instance
    
    def create_bucket_and_upload(self, bucket_name, file_path):
        """Create S3 bucket and upload a file"""
        print(f"Creating bucket {bucket_name}...")
        
        try:
            self.s3.create_bucket(Bucket=bucket_name)
        except Exception as e:
            print(f"Bucket might already exist: {e}")
        
        print(f"Uploading {file_path}...")
        self.s3.Bucket(bucket_name).upload_file(
            Filename=file_path,
            Key=file_path.split('/')[-1]
        )
        print("Upload complete")
        
    def create_monitoring_user(self, username):
        """Create IAM user with CloudWatch read permissions"""
        print(f"Creating user {username}...")
        
        try:
            self.iam.create_user(UserName=username)
        except self.iam.exceptions.EntityAlreadyExistsException:
            print("User already exists")
        
        policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "cloudwatch:GetMetricStatistics",
                        "cloudwatch:ListMetrics"
                    ],
                    "Resource": "*"
                }
            ]
        }
        
        policy_name = 'CloudWatchReadOnly'
        try:
            policy_response = self.iam.create_policy(
                PolicyName=policy_name,
                PolicyDocument=json.dumps(policy)
            )
            policy_arn = policy_response['Policy']['Arn']
        except self.iam.exceptions.EntityAlreadyExistsException:
            account_id = boto3.client('sts').get_caller_identity()['Account']
            policy_arn = f"arn:aws:iam::{account_id}:policy/{policy_name}"
        
        self.iam.attach_user_policy(
            UserName=username,
            PolicyArn=policy_arn
        )
        print(f"User {username} created with CloudWatch permissions")
        
    def setup_alarm(self, instance_id, email):
        """Create CloudWatch alarm for instance"""
        print("Setting up CPU alarm...")
        
        topic_response = self.sns.create_topic(Name='instance-alerts')
        topic_arn = topic_response['TopicArn']
        
        self.sns.subscribe(
            TopicArn=topic_arn,
            Protocol='email',
            Endpoint=email
        )
        
        self.cloudwatch.put_metric_alarm(
            AlarmName=f'HighCPU-{instance_id}',
            ComparisonOperator='GreaterThanThreshold',
            EvaluationPeriods=2,
            MetricName='CPUUtilization',
            Namespace='AWS/EC2',
            Period=300,
            Statistic='Average',
            Threshold=80.0,
            ActionsEnabled=True,
            AlarmActions=[topic_arn],
            Dimensions=[
                {
                    'Name': 'InstanceId',
                    'Value': instance_id
                }
            ]
        )
        print(f"Alarm created. Confirm email subscription at {email}")
        
    def run_full_setup(self):
        """Execute complete infrastructure setup"""
        print("Starting full AWS automation...")
        print("=" * 50)
        
        # Launch instance
        instance = self.launch_instance('AutomatedServer')
        
        # Create storage
        bucket_name = f'automation-demo-{int(time.time())}'
        self.create_bucket_and_upload(bucket_name, 'config.json')
        
        # Create monitoring user
        self.create_monitoring_user('monitoring-bot')
        
        # Setup alerts
        self.setup_alarm(instance.id, 'alerts@example.com')
        
        print("=" * 50)
        print("Setup complete!")
        print(f"Instance ID: {instance.id}")
        print(f"S3 Bucket: {bucket_name}")

if __name__ == '__main__':
    automation = AWSAutomation(region='us-east-1')
    automation.run_full_setup()

Save this as aws_automation.py and run it with python3 aws_automation.py. The script creates a complete environment in minutes.

You can extend this script to handle multiple instances, configure load balancers, set up databases, or manage VPCs. The pattern stays the same: create a method for each task, use Boto3 clients or resources, and handle errors appropriately.

Common Pitfalls

Here are mistakes to avoid when working with Boto3.

Hardcoded credentials: Never put access keys directly in code. Use environment variables, the credentials file, or IAM roles. If you commit credentials to Git, rotate them immediately.

# Bad
boto3.client('s3', 
    aws_access_key_id='AKIAIOSFODNN7EXAMPLE',
    aws_secret_access_key='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'
)

# Good
boto3.client('s3')  # Uses credentials from ~/.aws/credentials or IAM role

Wrong region: AWS resources live in specific regions. If you create an EC2 instance in us-west-2 but query us-east-1, you won’t find it. Always specify regions explicitly:

ec2 = boto3.resource('ec2', region_name='us-west-2')

Not handling pagination: Many AWS API calls return paginated results. The instances.all() method handles this automatically, but when using clients, you need paginators:

# Wrong - only gets first page
s3_client = boto3.client('s3')
response = s3_client.list_objects_v2(Bucket='my-bucket')
for obj in response.get('Contents', []):
    print(obj['Key'])

# Correct - gets all pages
paginator = s3_client.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket='my-bucket'):
    for obj in page.get('Contents', []):
        print(obj['Key'])

Ignoring errors: AWS operations can fail for many reasons (permissions, quotas, network issues). Always wrap API calls in try-except blocks:

try:
    instance = ec2.Instance('i-1234567890abcdef0')
    instance.terminate()
except boto3.exceptions.Boto3Error as e:
    print(f"Failed to terminate instance: {e}")

Forgetting to clean up: Running instances and storing data costs money. Always delete resources you no longer need:

# Tag resources for automatic cleanup
instances = ec2.create_instances(
    ImageId='ami-0c55b159cbfafe1f0',
    MinCount=1,
    MaxCount=1,
    InstanceType='t2.micro',
    TagSpecifications=[
        {
            'ResourceType': 'instance',
            'Tags': [
                {'Key': 'Environment', 'Value': 'Testing'},
                {'Key': 'AutoDelete', 'Value': 'true'}
            ]
        }
    ]
)

Then create a Lambda function that runs daily and terminates instances tagged with AutoDelete: true.

Missing IAM permissions: Your credentials need appropriate permissions. If operations fail with access denied errors, check your IAM policies. Start with broader permissions during development, then narrow them for production.

Summary

Boto3 gives you full control over AWS infrastructure through Python code. You’ve learned to manage EC2 instances, handle S3 storage, configure IAM users, and set up CloudWatch monitoring. The complete automation script shows how to combine these operations into a single workflow.

Start with small tasks like listing resources or uploading files. Once you’re comfortable, build larger automation scripts. Consider adding error handling, logging, and configuration files to make your scripts production-ready.

The AWS API covers hundreds of services beyond what we covered here. Lambda, RDS, DynamoDB, ECS, and many others all have Boto3 support. The patterns you learned here apply to all of them: create a client or resource, call methods, handle responses and errors.

Check the Boto3 documentation for complete API references. The AWS SDK examples repository contains hundreds of code samples for different services.

Write scripts for tasks you do repeatedly. Automate backups, rotate logs, scale infrastructure, or deploy applications. Boto3 saves time and reduces errors compared to manual operations.