Are you struggling to navigate through the AWS SDK for Python Boto3 documentation? Do you want to speed up your development process with pre-written code snippets? Look no further!

Boto3 Python AWS SDK 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

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())

Also see: How To Send Transactional Emails Using Lambda and Amazon SES?

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 to 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,
    [email protected],
    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 an 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.

Do not hesitate to contact us for assistance with your cybersecurity needs. Our team of experts is here to provide top-notch support and ensure the safety and security of your data and systems. Check out our DevSecOps services to stay secure!