AWS SDK (Boto3) for Python: Cloud Infrastructure Automation

Boto3 is the AWS SDK for Python, enabling developers to write software that uses AWS services programmatically. It provides object-oriented APIs and low-level access to AWS services.

Installation and Configuration

Install Boto3

pip install boto3

Configure AWS Credentials

# Configure credentials
aws configure

# Or set environment variables
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_DEFAULT_REGION=us-east-1

Create Boto3 Client

import boto3

# Create client
s3_client = boto3.client('s3', region_name='us-east-1')

# Create resource (higher-level interface)
s3_resource = boto3.resource('s3')

# Create session for multiple services
session = boto3.Session(
    aws_access_key_id='YOUR_ACCESS_KEY',
    aws_secret_access_key='YOUR_SECRET_KEY',
    region_name='us-east-1'
)

ec2_client = session.client('ec2')
s3_client = session.client('s3')

S3 (Simple Storage Service)

Bucket Operations

import boto3

s3_client = boto3.client('s3')

# Create bucket
s3_client.create_bucket(Bucket='my-bucket')

# List buckets
response = s3_client.list_buckets()
for bucket in response['Buckets']:
    print(f"Bucket: {bucket['Name']}")

# Delete bucket
s3_client.delete_bucket(Bucket='my-bucket')

# Check if bucket exists
try:
    s3_client.head_bucket(Bucket='my-bucket')
    print("Bucket exists")
except:
    print("Bucket does not exist")

Object Operations

# Upload file
s3_client.upload_file(
    Filename='local_file.txt',
    Bucket='my-bucket',
    Key='remote_file.txt'
)

# Upload with metadata
s3_client.put_object(
    Bucket='my-bucket',
    Key='file.txt',
    Body=b'file content',
    Metadata={'author': 'john', 'version': '1.0'}
)

# Download file
s3_client.download_file(
    Bucket='my-bucket',
    Key='remote_file.txt',
    Filename='local_file.txt'
)

# Get object
response = s3_client.get_object(Bucket='my-bucket', Key='file.txt')
content = response['Body'].read()

# List objects
response = s3_client.list_objects_v2(Bucket='my-bucket')
for obj in response.get('Contents', []):
    print(f"Object: {obj['Key']}, Size: {obj['Size']}")

# Delete object
s3_client.delete_object(Bucket='my-bucket', Key='file.txt')

# Copy object
s3_client.copy_object(
    Bucket='my-bucket',
    CopySource={'Bucket': 'source-bucket', 'Key': 'source-file.txt'},
    Key='destination-file.txt'
)

S3 Resource Interface

s3 = boto3.resource('s3')

# Access bucket
bucket = s3.Bucket('my-bucket')

# Upload file
bucket.upload_file('local_file.txt', 'remote_file.txt')

# Download file
bucket.download_file('remote_file.txt', 'local_file.txt')

# List objects
for obj in bucket.objects.all():
    print(f"Object: {obj.key}, Size: {obj.size}")

# Delete object
obj = bucket.Object('file.txt')
obj.delete()

EC2 (Elastic Compute Cloud)

Instance Management

ec2_client = boto3.client('ec2')

# Launch instance
response = ec2_client.run_instances(
    ImageId='ami-0c55b159cbfafe1f0',  # Amazon Linux 2
    MinCount=1,
    MaxCount=1,
    InstanceType='t2.micro',
    KeyName='my-key-pair',
    SecurityGroups=['default'],
    TagSpecifications=[
        {
            'ResourceType': 'instance',
            'Tags': [
                {'Key': 'Name', 'Value': 'my-instance'},
                {'Key': 'Environment', 'Value': 'dev'}
            ]
        }
    ]
)

instance_id = response['Instances'][0]['InstanceId']
print(f"Instance ID: {instance_id}")

# Describe instances
response = ec2_client.describe_instances()
for reservation in response['Reservations']:
    for instance in reservation['Instances']:
        print(f"Instance: {instance['InstanceId']}, State: {instance['State']['Name']}")

# Stop instance
ec2_client.stop_instances(InstanceIds=[instance_id])

# Start instance
ec2_client.start_instances(InstanceIds=[instance_id])

# Terminate instance
ec2_client.terminate_instances(InstanceIds=[instance_id])

# Reboot instance
ec2_client.reboot_instances(InstanceIds=[instance_id])

Security Groups

# Create security group
response = ec2_client.create_security_group(
    GroupName='my-sg',
    Description='My security group',
    VpcId='vpc-12345678'
)

sg_id = response['GroupId']

# Add ingress rule
ec2_client.authorize_security_group_ingress(
    GroupId=sg_id,
    IpPermissions=[
        {
            'IpProtocol': 'tcp',
            'FromPort': 80,
            'ToPort': 80,
            'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
        },
        {
            'IpProtocol': 'tcp',
            'FromPort': 443,
            'ToPort': 443,
            'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
        }
    ]
)

# Describe security groups
response = ec2_client.describe_security_groups()
for sg in response['SecurityGroups']:
    print(f"Security Group: {sg['GroupName']}, ID: {sg['GroupId']}")

Lambda

Create and Invoke Functions

lambda_client = boto3.client('lambda')

# Create function
with open('lambda_function.zip', 'rb') as f:
    zip_content = f.read()

response = lambda_client.create_function(
    FunctionName='my-function',
    Runtime='python3.11',
    Role='arn:aws:iam::123456789012:role/lambda-role',
    Handler='index.handler',
    Code={'ZipFile': zip_content},
    Description='My Lambda function',
    Timeout=30,
    MemorySize=128,
    Environment={
        'Variables': {
            'ENV': 'production',
            'DEBUG': 'false'
        }
    }
)

# Invoke function
response = lambda_client.invoke(
    FunctionName='my-function',
    InvocationType='RequestResponse',  # Synchronous
    Payload=b'{"key": "value"}'
)

result = json.loads(response['Payload'].read())
print(f"Result: {result}")

# Async invocation
lambda_client.invoke(
    FunctionName='my-function',
    InvocationType='Event',  # Asynchronous
    Payload=b'{"key": "value"}'
)

# List functions
response = lambda_client.list_functions()
for func in response['Functions']:
    print(f"Function: {func['FunctionName']}, Runtime: {func['Runtime']}")

# Update function code
with open('lambda_function.zip', 'rb') as f:
    zip_content = f.read()

lambda_client.update_function_code(
    FunctionName='my-function',
    ZipFile=zip_content
)

# Delete function
lambda_client.delete_function(FunctionName='my-function')

DynamoDB

Table Operations

dynamodb = boto3.resource('dynamodb')

# Create table
table = dynamodb.create_table(
    TableName='users',
    KeySchema=[
        {'AttributeName': 'user_id', 'KeyType': 'HASH'},
        {'AttributeName': 'timestamp', 'KeyType': 'RANGE'}
    ],
    AttributeDefinitions=[
        {'AttributeName': 'user_id', 'AttributeType': 'S'},
        {'AttributeName': 'timestamp', 'AttributeType': 'N'}
    ],
    BillingMode='PAY_PER_REQUEST'
)

# Put item
table.put_item(
    Item={
        'user_id': 'user123',
        'timestamp': 1234567890,
        'name': 'John Doe',
        'email': '[email protected]'
    }
)

# Get item
response = table.get_item(
    Key={
        'user_id': 'user123',
        'timestamp': 1234567890
    }
)
item = response['Item']

# Query items
response = table.query(
    KeyConditionExpression='user_id = :uid',
    ExpressionAttributeValues={':uid': 'user123'}
)

for item in response['Items']:
    print(f"Item: {item}")

# Scan table
response = table.scan()
for item in response['Items']:
    print(f"Item: {item}")

# Update item
table.update_item(
    Key={'user_id': 'user123', 'timestamp': 1234567890},
    UpdateExpression='SET #name = :name, #email = :email',
    ExpressionAttributeNames={'#name': 'name', '#email': 'email'},
    ExpressionAttributeValues={':name': 'Jane Doe', ':email': '[email protected]'}
)

# Delete item
table.delete_item(
    Key={'user_id': 'user123', 'timestamp': 1234567890}
)

# Delete table
table.delete()

RDS (Relational Database Service)

rds_client = boto3.client('rds')

# Create DB instance
response = rds_client.create_db_instance(
    DBInstanceIdentifier='my-database',
    DBInstanceClass='db.t3.micro',
    Engine='postgres',
    MasterUsername='admin',
    MasterUserPassword='password123',
    AllocatedStorage=20,
    StorageType='gp2'
)

# Describe DB instances
response = rds_client.describe_db_instances()
for db in response['DBInstances']:
    print(f"DB: {db['DBInstanceIdentifier']}, Status: {db['DBInstanceStatus']}")

# Modify DB instance
rds_client.modify_db_instance(
    DBInstanceIdentifier='my-database',
    AllocatedStorage=30,
    ApplyImmediately=True
)

# Create DB snapshot
rds_client.create_db_snapshot(
    DBSnapshotIdentifier='my-snapshot',
    DBInstanceIdentifier='my-database'
)

# Delete DB instance
rds_client.delete_db_instance(
    DBInstanceIdentifier='my-database',
    SkipFinalSnapshot=True
)

Error Handling

from botocore.exceptions import ClientError

try:
    s3_client.get_object(Bucket='my-bucket', Key='nonexistent.txt')
except ClientError as e:
    error_code = e.response['Error']['Code']
    if error_code == 'NoSuchKey':
        print("Object not found")
    elif error_code == 'NoSuchBucket':
        print("Bucket not found")
    else:
        print(f"Error: {error_code}")
except Exception as e:
    print(f"Unexpected error: {e}")

Best Practices

Use IAM roles: Avoid hardcoding credentials
Error handling: Handle ClientError exceptions
Pagination: Use paginator for large result sets
Resource cleanup: Use context managers or finally blocks
Logging: Log API calls for debugging
Retry logic: Implement exponential backoff
Cost optimization: Monitor API calls and optimize

Common Pitfalls

Bad Practice:

# Don't: Hardcode credentials
s3 = boto3.client(
    's3',
    aws_access_key_id='AKIAIOSFODNN7EXAMPLE',
    aws_secret_access_key='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'
)

# Don't: No error handling
s3.get_object(Bucket='bucket', Key='key')

# Don't: Inefficient pagination
response = s3.list_objects_v2(Bucket='bucket')
all_objects = response['Contents']

Good Practice:

# Do: Use IAM roles or environment variables
s3 = boto3.client('s3')

# Do: Handle errors
try:
    s3.get_object(Bucket='bucket', Key='key')
except ClientError as e:
    print(f"Error: {e}")

# Do: Use paginator
paginator = s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket='bucket'):
    for obj in page.get('Contents', []):
        print(obj['Key'])

Conclusion

Boto3 provides comprehensive access to AWS services from Python. Master core services like S3, EC2, Lambda, and DynamoDB to build scalable cloud applications. Use proper error handling, IAM roles, and best practices to build secure, maintainable infrastructure automation.