Aug 3rd, 2021

AWS SDK for Python: Boto3 Cheat Sheet & Code Snippets

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.

Boto3 installation

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

Importing Boto3 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

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

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

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

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

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)

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.

If you liked this post, share it now!

Our Recent Posts

PM and EDR Remediation Demonstration

Explore how to patch using Qualys PM and remediate the target host with Qualys EDR modules. Fin...

Read More

Qualys Endpoint Detection and Response (EDR)

Learn about Endpoint Detection and Response (EDR) essentials and its benefits. Follow the Qualy...

Read More

Qualys Patch Management (PM)

Need a guide for Qualys PM? Explore the Qualys PM application, assets and patches. Follow the P...

Read More