AWS SDK for Python Boto3 Cheat Sheet & Code Snippets

August 03, 2021



AWS SDK for Python - Boto3 features with code snippets. Boto3 Cheat Sheet, aws sdk for python, aws boto3 python documentation, aws boto3 dynamodb, boto3 s3, boto3 create prefix, boto3 prefix

Overview

Boto3 is the AWS SDK(Software Development Kit) for Python. Boto3 comes with many functionalities that may seem confusing at the first glance.

This blog post aims to ease the use of Boto3 by providing the most used functionalities, as well as some extra code snippets.

Boto 3 installation

pip install boto3
# or you can use the below statement
python3 -m pip install boto3

Importing Boto 3 and creating a client the right way

It’s not recommended to hard-code your AWS credentials in code like in the below example.

# NON-RECOMMENDED WAY
import boto3
client = boto3.client(
    s3,
    aws_access_key_id=yyyy,     # This is not recommended and not secure
    aws_secret_access_key=xxxx, # Please configure ~/.aws/credentials file
    region_name=us-east-1       # and boto3 will get the credentials from there.
)
# RECOMMENDED WAY
import boto3
client = boto3.client(
    s3,                         # boto3 will get the credentials either from
    region_name=us-east-1       # environment vars. or the .aws/credentials file.
)

S3 - Simple Storage Service

S3 - Simple Storage Service

a. Upload in-memory files to S3

This snippet lets you download an image from the internet and upload it to S3 without saving the image to the filesystem.

import boto3, requests # you can `pip install requests`
from io import BytesIO

s3 = boto3.resource(s3)
bucket = s3.Bucket(your-bucket-name)

image_url = "http://www._____.jpg"
image_content = requests.get(image_url).content # get the images content
img = BytesIO(image_content)
content_type = f"image/{file_format}"
response = bucket.put_object(Key=images/testimage.jpg, Body=img.getvalue(), ContentType="image/jpg")
uploaded_bucket, uploaded_key = response.bucket_name, response.key

b. Delete S3 objects filtered by prefix

This snippet lets you filter objects with prefixes from S3 and delete them. Prefixes work in the same way that str.startswith() does.

s3 = boto3.resource(s3)
bucket = s3.Bucket(your-bucket-name)
objects = bucket.objects.filter(Prefix=photos/year/{}.format(2021))
objects.delete()

Lambda Functions

Lambda Functions

a. [Synchronously] Invoke a Lambda Function

Normal invocation of a Lambda Function. The RequestResponse invocation type waits for execution to be complete.

client = boto3.client(lambda)
result = client.invoke(
    FunctionName=arn:..........,
    InvocationType=RequestResponse,
    Payload=json.dumps({data:data}),
    LogType=Tail
)
response = json.loads(result[Payload].read())

b. [Asynchronously] Invoke a Lambda Function

Lambda Function gets executed, but it doesn’t wait for it to finish, so no response data is returned.

client = boto3.client(lambda)
result = client.invoke(
    FunctionName=arn:..........,
    InvocationType=Event,
    Payload=json.dumps({data:data}),
    LogType=Tail
)
response = json.loads(result[Payload].read())

SQS Simple Queue Service

SQS - Simple Queue Service

a. List SQS queues

This snippet lets you list the existing SQS queues, optionally with prefix filtering.


client = boto3.client(sqs)
response = client.list_queues(
    QueueNamePrefix=your-, # delete this if you want them all
)

b. Send SQS message

This snippet lets you add a new message to your SQS queue.


client = boto3.client(sqs)
queue = client.get_queue_by_name(QueueName=your-queue-name-goes-here)
response = queue.send_message(
    MessageBody=Example SQS Message content.,
    MessageAttributes={
                        custom_msg_tags: {
                            StringValue: are_supported,
                            DataType: String
                        }
                    },
)

c. Recieve a single SQS message

This snippet lets you receive a number of (denoted with MaxNumberOfMessages) messages from the SQS queue.

client = boto3.client(sqs)
queue = client.get_queue_by_name(QueueName=your-queue-name-goes-here)
response = queue.receive_message(
    QueueUrl=self.queue_url,
    MessageAttributeNames=[All],
    MaxNumberOfMessages=1,
)

SNS Simple Notification Service

SNS - Simple Notification Service

a. Creating a new SNS Topic & Subscribing an email

This snippet creates an SNS Topic and creates a subscription for an email to the SNS Topic.

import boto3
client = boto3.client(sns)
# creating a new topic
create_topic_response = client.create_topic(
    Name=your-topic-name
)
created_topic_arn = create_topic_response.get(TopicArn)

# subscribing an email to the topic
subscribe_response = client.subscribe(
    TopicArn=created_topic_arn,
    Protocol=email,
    Endpoint=some@address.com,
    ReturnSubscriptionArn=True
)
created_subscription_arn = create_topic_response.get(SubscriptionArn)

b. Publishing a new message to SNS Topic

This snippet lets you publish a new SNS message which will be delivered to all subscribers.

import boto3
client = boto3.client(sns)
response = client.publish(
    TopicArn=arn:......:your-topic-name,
    Message=Your SNS message that will be distributed to subscribers.,
    Subject=Your Emails Subject
)

Cloud Watch

Cloud Watch

Create a EventBridge rule and a Lambda Target

This snippet lets you create a fixed schedule EventBridge Rule in CloudWatch and attaches a Lambda Function to said event.

import boto3


# Create CloudWatchEvents client
cloudwatch = boto3.client(events)
rule_name = EventBridgeRule
# Put an event rule
response = cloudwatch.put_rule(
    Name=rule_name,
    ScheduleExpression=rate(15 minutes),
    State=ENABLED
)
print(response)

# Put lambda target for rule
target_response = cloudwatch.put_targets(
    Rule=rule_name,
    Targets=[
        {
            Arn: LAMBDA_FUNCTION_ARN,
            Id: myCloudWatchEventsTarget,
        }
    ]
)
print(target_response)

EC 2

EC2

a. Delete unused EBS volumes

This snippet lets you delete the unattached EBS Volumes and prints out deleted volumes’ information.

ec2 = boto3.resource(ec2, region_name=us-east-1)

# List only unattached volumes (available vs. in-use)
volumes = ec2.volumes.filter(
    Filters=[{Name: status, Values: [available]}])

for volume in volumes:
    v = ec2.Volume(volume.id)
    v.delete()
    print("EBS volume: {}, Size: {} GiB is deleted".format(v.id, v.size))

b. List running EC2 instances

This snippet lets you list the existing and running EC2 instances.

client = boto3.client("ec2", region_name="us-east-1")
reservations = client.describe_instances(Filters=[{
        "Name": "instance-state-name",
        "Values": ["running"],
}]).get("Reservations")

instances = [r.get("Instances") for r in reservations]

Conclusion

In this blog post, we provided some snippets to make the use of Boto3 easier. We hope you find it helpful whether you’re new to Boto3 or in case you were looking for some Boto3 snippets.