Writing REST APIs

Beginning CRUD (Create, Read, Update, Delete) Operations

Hello everyone, today I will be working on APIs basically focusing in CRUD operations.

Source code is available on Github.

Create (C)

We are going to create an Organization directory and as we know, every organization has users or employees. Thus, we are going to perform four different operations like Create, Read, Update, Delete.

Let me start creating the organizations directory. But before that let me delete the hello world.

crud_step_1

I am going to place requirements.txt under organizations. Basically it needs to be placed under the users directory. I just forgot to do that, but I have rectified in my upcoming video. Don’t worry, you can find the entire tutorial source code in Github.

crud_step_2

Let me give you brief idea about the packages which I am using in this project.

requests package allows you to send HTTP requests using Python.

ujson is also known as ultra json and written purely in C. It is very fast compared to the standard json module.

pymongo & dnspython both are required dependencies for connecting to a MongoDB Database.

marshmallow module helps convert complex data types, such as objects, to and from native Python data types. This module is normally used with Flask but if you are coming from Django Rest Framework background you can compare it with serializers.

pyjwt is a Python library which allows you to encode and decode JSON Web Tokens (JWT). We will be using the JWT module in the Login & Refresh Token APIs.

argon2 is a cryptographic hashing algorithm, most recommended for password hashing. It is also the winner of the 2015 Password Hashing Competition, a community-organized open competition to select a next generation hashing algorithm. It’s designed not to be easier to compute on custom hardware than it is to compute on an ordinary CPU.

  • Make sure to run pip install -r requirements.txt

I will be creating different packages like (create,read, update, delete) under users.

Under the create directory, I will create an app.py file and utils package where I will be storing basic utilities like db connection and validator.

Now, I am going to write the code for connecting to MongoDB, but before that let me open MongoDB Atlas and copy the connection string for Python driver.

crud_step_3

I will be creating a MongoDBConnection Class. This class is actually a context manager, as you may know Context managers allow you to allocate and release resources precisely when you want to. The primary motivation behind context managers is resource management.

As you can see in the below code, we have used special methods __enter__ and __exit__. The enter method will help to establish connection with the database, and the exit method will take care to close the connection.

from pymongo import MongoClient
class MongoDBConnection:
def __init__(self):
self.connection = None
def __enter__(self):
self.connection = MongoClient(
"mongodb+srv://<USERNAME>:<PASSWORD>@<DBHOST>/<DB_NAME>")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.connection.close()

Now I will open the validator.py file and start creating a schema using the marshmallow module.

We won’t be storing plain-text passwords in the database. You can see that we are using the encrypt function which returns the hashed value.

from marshmallow import Schema, fields, post_load, ValidationError
from argon2 import PasswordHasher
from . import db
def encrypt(plain_text_password):
""" This function takes plain-text password and returns encrypted password """
ph = PasswordHasher()
hashed_password = ph.hash(plain_text_password)
return hashed_password
class UserRegistrationSchema(Schema):
first_name = fields.Str(required=True)
last_name = fields.Str(required=True)
email = fields.Email(required=True)
password = fields.Str(required=True)
@post_load
def encrypt_password(self, data, **kwargs):
data["password"] = encrypt(data["password"])
return data
@post_load
def validate_email(self, data, **kwargs):
mongo = db.MongoDBConnection()
with mongo:
database = mongo.connection["myDB"]
collection = database["registrations"]
# check email already exists in DB. If true
# raise validation error.
if collection.find_one({"email": data["email"]}) is not None:
raise ValidationError('This email address is already taken.')
return data

You can see how the schema looks similar to DRF Serializers. For post-processing you can see we are using post_load decorator for encrypting our passwords as well validating email for their existence in the db.

Now, I am going to open app.py where the main business logic resides.

Let me first import the necessary modules. Next, I will define the lambda handler.

crud_step_4

All the information will be coming under the event body. I need to parse the information using the json module.

The UserRegistrationSchema will be validating and processing whether all the inputs provided are correct or not. If the validation fails, or there is a generic key error, then it will raise a bad request, otherwise I will store the record in the registrations collection and return 201 HTTP response.

Below code is the final snippet how the app.py is going to look like.

import ujson
from marshmallow import ValidationError
from .utils import db, validator
def lambda_handler(event, context):
try:
body = ujson.loads(event["body"])
result = validator.UserRegistrationSchema()
# Check if dictionary is empty.
res = not bool(result.validate(body))
if res:
# Store information in DB
mongo = db.MongoDBConnection()
with mongo:
database = mongo.connection["myDB"]
collection = database["registrations"]
collection.insert_one(result.load(body))
return {
"statusCode": 201,
"body": ujson.dumps({
"message": "Registered Successfully",
"data": result.validate(body)
})
}
else:
return {
"statusCode": 400,
"body": ujson.dumps({"message": "Error !",
"data": result.validate(body)
})
}
except ValidationError as err:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": err.messages
})
}
except KeyError:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": "Something went wrong. Unable to parse data !"
})
}

Now, I am going to open the template.yaml file.

I have a global timeout of 30 seconds, and set the memory size to 2GB, it's completely based on your preference.

crud_step_5

After setting the memory limit, I am going to register the API.

crud_step_6

Codeuri points to the directory where my code resides and Handler is combination of folder_name.file_name.function_name

crud_step_7

Please make sure requirements.txt is placed under the user directory, otherwise the build will fail.

We are now ready to test our application. Let me copy a sample snippet from event.json. I will be passing dummy information in the body to check whether my function is able to create a new user.

crud_step_8

Okay, as you can see in the console, the user has been registered successfully.

crud_step_9

Let me verify the data is also present in MongoDB.

Yes, the record has been successfully inserted in our database and even the password is stored in encrypted format.

crud_step_10

Finally, we can store the user information in our database. Let’s move and create an API which retrieves a list of users from the database.

Read (R)

I am going to create a read package under the user. I will follow a similar kind of implementation as I did for the create operation.

I am going to copy the db.py file from utils and place it under the read/utils.

crud_step_11

Same as previous, I will create app.py where I will be defining my business logic.

You can see on line 7 that I am trying to retrieve the id from path parameters. The ID is referring to the primary key of a specific user.

crud_step_12

The retrieve_info function will take object_id as an input. Object_id is referring to the ID which we receive from PathParameters, if object_id is present in the db we return information related to that specific user else we return all users list.

This is how the final code going to look like.

import bson
import ujson
from bson import ObjectId
from .utils import db
def retrieve_info(object_id):
# Retrieve info from DB
result = list()
mongo = db.MongoDBConnection()
with mongo:
database = mongo.connection['myDB']
collection = database['registrations']
if object_id is not None:
# retrieve particular object from db.
try:
single_object = collection.find_one({"_id": ObjectId(object_id)})
if single_object is None:
return {"error": "The respective id does not exist !"}
except bson.errors.InvalidId:
return {"error": "Please provide a valid ObjectId"}
return {
'id': object_id,
'first_name': single_object["first_name"],
'last_name': single_object["last_name"],
'email': single_object["email"]
}
else:
# retrieve all information from db.
for data in collection.find():
result.append({
"id": str(data["_id"]),
"first_name": data["first_name"],
"last_name": data["last_name"],
"email": data["email"]
})
return result
def lambda_handler(event, context):
try:
object_id = event["pathParameters"]["Id"]
except TypeError:
object_id = None
except KeyError:
object_id = None
try:
return {
"statusCode": 200,
"body": ujson.dumps({
"message": "Success",
"data": retrieve_info(object_id)
})
}
except Exception as err:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": "Something went wrong. Unable to parse data !",
"error": str(err)
})
}

The implementation is done. Let’s move to template.yaml file and register our API.

We will register two APIs. One is OrganizationUserRead and OrganizationUserReadById.

crud_step_13

crud_step_14

You can see both the APIs are pointing to the same business logic. In ReadById we are passing Id as a parameter for getting information for a particular user.

Let’s test out the functionality. I will click on Run and then Edit Configurations. This is a GET API we don’t need to pass anything in the body.

crud_step_15

On a Windows machine you will get a Docker prompt for file sharing, kindly accept it.

As you can below in the console, we are receiving all users list.

crud_step_16

Now, I will try to retrieve information for a specific user. Let me first go to the registrations collection.

I will copy the _id for the user Mike Jones.

crud_step_17

I will then paste the _id it in the pathParameters. I will click on Apply and then OK.

crud_step_18

Before running the app, I have found an issue which needs to be fixed otherwise our code is not going work.

Open app.py goto line number 45 it should be “Id”, I should be uppercase.

crud_step_19

Save it and let me try to run again.

crud_step_20

It's working fine. We are getting information for user Mike Jones.

Finally, our API is able to retrieve information from the database. Let’s move on and create an API which will be used to update records in our database.

Update (U)

I am going to create an update package under the user. I will follow the same approach that I did before for create & read operations.

As you can see over here at line 10 that we are again capturing Id. If the Id is present in the database we will move forward and update all the required fields else it will raise an error.

crud_step_21

I will call the update_one operation, observe line number 25 which will search whether the ObjectId exists in the database if yes then it will set the new values. If the update is successful then it will return 200 HTTP Response.

crud_step_22

This is how the final code for the update operation is going to look like.

import bson
import ujson
from bson import ObjectId
from .utils import db, validator
def lambda_handler(event, context):
try:
object_id = event["pathParameters"]["Id"]
except TypeError:
object_id = None
body = ujson.loads(event['body'])
result = validator.UserSchema()
res = not bool(result.validate(body))
if res:
mongo = db.MongoDBConnection()
with mongo:
database = mongo.connection['myDB']
collection = database['registrations']
try:
collection.update_one({"_id": ObjectId(object_id)}, {"$set": body})
except bson.errors.InvalidId:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": "Error ! Please provide a valid ObjectId",
"data": None
})
}
return {
"statusCode": 200,
"body": ujson.dumps({
"message": "Data Updated Successfully !",
"data": result.dump(body)
})
}
else:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": "Error !",
"data": result.validate(body)
})
}

Let’s register the API in the template.yaml file. As this API is being used to update our records, so the method is going to be PUT.

crud_step_23

Now, let’s test the functionality. I will click on Run and then Edit Configurations.

I will pass necessary information like id in path parameters and in the body I will send the values for firstname, lastname and email which needs to be updated.

crud_step_24

Apply the changes and try to run again.

Finally, our data has been successfully updated.

crud_step_25

Let me refresh the registrations collection and verify whether data has been updated or not.

crud_step_26

Great ! The data has been updated.

Delete (D)

Let’s now move to the final part where we are going to remove records from the database by creating a DeleteAPI.

I am going to perform the same kind of operations that I did previously for other operations.

You can see over here at line 9 that we are again capturing Id. I will call the delete_one operation which will remove the object from the database.

crud_step_27

We are performing a similar kind of operation what we did for the update.

Okay, our implementation is done. This is how the final code is going to look like.

import bson
import ujson
from bson import ObjectId
from .utils import db
def lambda_handler(event, context):
try:
object_id = event["pathParameters"]["Id"]
except TypeError:
object_id = None
mongo = db.MongoDBConnection()
with mongo:
database = mongo.connection['myDB']
collection = database['registrations']
try:
collection.delete_one({"_id": ObjectId(object_id)})
except bson.errors.InvalidId:
return {
"statusCode": 400,
"body": ujson.dumps({
"message": "Error ! Invalid ObjectId",
"data": None
})
}
return {
"statusCode": 204,
"body": ujson.dumps({
"message": "Data Deleted !",
"data": None
})
}

Let’s now register our API in template.yaml file and test out the functionality.

I am going to remove the last row David Jones from the registrations collection.

Click on Run and then Edit Configurations. I will pass the id of David Jones in pathParameters.

crud_step_28

I will apply the changes and try to run again.

The data has been successfully deleted from the db. Let me refresh the collection and verify it.

crud_step_29

Great! The record has been removed.

crud_step_30

We have successfully completed the entire CRUD (Create, Read, Update & Delete) APIs. In the upcoming tutorial we will be working on the Custom Authorizers to protect our APIs.