Building a Real-Time Tracking and Shipment Notification System - using AWS SQS, Lambda, and more

Scenario:

A company wants to create a system that can track customer orders and send notifications when orders have been shipped. They want to use AWS services to build the system. They have decided to use SQS, Lambda, and Python for the project.

Overview

In this project, I created a robust system that tracks customer orders and sends timely notifications upon shipment using AWS services. Specifically, AWS Simple Queue Service (SQS) and Lambda, as well as Python.

Creating the SQS queue

The following code snippet demonstrates how I created the SQS queue by referring to the official Boto3 documentation. I did this in my Cloud9 environment. If you are unsure of how to set up your Cloud9 environment, take a look at one of my previous projects.

File Navigator: Simplifying File Information Retrieval with Python Scripting

Each line of code is accompanied by inlined comments explaining its purpose.

import boto3

# This gets the service resource
sqs = boto3.resource('sqs')

# This creates the queue and name
queue = sqs.create_queue(QueueName='customer_orders')

# This prints the url
print(queue.url)

Now when I open the SQS page in the AWS console, I can see the queue just created.

Creating the Lambda function

I created the function by authoring from scratch. I chose the runtime python 3.10, and kept the architecture at x86_64. For the execution role, I will be creating a new role with basic lambda permissions.

After selecting the Lambda as the AWS service, I then attached the appropriate permission policy.

This policy gives SQS full access.

Now that the role is created, I will now return to my lambda function and attach it as a existing role.

The lambda function is now all created!

Modifying the Lambda function

Now I will be modifying the Lambda function to send a message to the SQS queue. For this project I’ll just be sending random generated numbers, I’ll do this by importing the random module.

To send the messages, I’ll be utilizing part of the code found in the boto3 documentation.

response = client.send_message(
    QueueUrl='string',
    MessageBody='string',
    DelaySeconds=123

and now to modify the code in the actual lambda function. I began by importing the modules I needed, boto3 and random. json was already imported.

I then edited the above code for this function, ie., I added in the queue url, and created the message to be a random number as a string. I also decided to change the number of delay seconds to 2.

Lastly I edited the code to return the message I created. Which would be a random generated number.

import json
import boto3
import random 


sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/456863550460/customer_orders'


def lambda_handler(event,context): 
    message = str(random.randint(1,10))
    response = sqs.send_message(
    QueueUrl= queue_url,
    MessageBody=message,
    DelaySeconds=2)


    return {
        'statusCode': 200,
        'body': json.dumps(message)
    }

Note: When writing out my code I bumped into a couple trip ups. Firstly, I mistakingly wrote 1–10 instead of 1,10 on the sixth line. This gave me a “TypeError: Random.randint() missing 1 required positional argument: ‘b’” error.

I also kept getting a “Hello from Lambda” as a response instead of my generated numbers. This was just because I forgot to deploy my changes before testing it.

I then went ahead and clicked “test”, created a name for it and chose SQS as a template.

Success!

Creating an API Gateway HTTP API type trigger

Now to set up the API gateway HTTP type trigger, This will allow us to invoke the Lambda function through an API endpoint.

I selected the HTTP API and then integrated the Lambda function I created earlier to it.

I edited the routes to default and then modified the integration target to the lambda function.

Lastly I kept the stages as is.

Testing the trigger

Now that everything has been set up and configured, all thats left is to test out if everything is working correctly.

In the API gateway page for the trigger just created, there is an invoke url. When invoked I should be getting random numbers.

when invoking this url, I got a 2,

When invoking again I got a different number,

and again,

It definitely works!

Now to check if the messages are being polled in the SQS,

Upon opening them up and cross referencing, they are all correct! We just created a Real-Time Tracking and Shipment Notification System!

I pushed all the code from my cloud9 IDE to my GitHub, Feel free to take a look! I have the function code I used as well as the SQS queue creation code commited!

GitHub - HafsahR22/python-repo: Repository for python exercises

Thank you so much for taking the time to read through my article, I hope you enjoyed it and found it insightful. Feel free to leave any questions or comments below. For more breakdowns and documentations of my projects, follow me!