Creating Simple REST APIv3 Using Python Flask, SQLite3, and Postman

بسم الله الرحمن الرحيم
احييكم بتحية الإسلام السلام عليكم ورحمة الله تعالى وبركاته

Hello world, Hossam is here, welcome to my 3rd blog in the Graduation Project Diary series, in this blog I try to explain and discuss new things that I have learned during the final year @ BFCAI ❤️. In this post I’ll try to explain how to implement Python Flask API for my project teammates to be aware of this topic and do their tasks 😃

Introduction

Lets start with building Simple CRUD API using Flask and SQLite3, our example project we will create a Books table that we can Create, Retrieve, Update, Delete (CRUD). i also will mention how to implement database API that provide you with simple way to interact with database in an easy, and readable way. Our project will handle requests coming from the user via HTTP, and then we will apply this to the database. lets start by creating project directory and Database. I’ll use this structure.

/API
	/cores
		__inti__.py
		database_api.py
		booksDB.db
	api.py

Building Database API

In this section I’ll build python class called DatabaseAPI to be my interface and formal way to interact with my SQlite3 database. in this approach our API/application code will be clean and free of SQL syntax because all database syntax and operation will be handled by DatabaseAPI which is good for encapsulation and abstraction. lets start with the __init__(self) method

Database Schema

building the Books Table with the following simple code

CREATE TABLE "Books" (
	"Id"	INTEGER UNIQUE,
	"title"	TEXT NOT NULL,
	"author"	TEXT NOT NULL,
	"description"	TEXT NOT NULL,
	"pages"	INTEGER NOT NULL,
	PRIMARY KEY("Id" AUTOINCREMENT)
);

Init method`

here in the __init__() method we can pass the database name to be connected with, in this approach i can connect to multiple database if i have to, then we created a method to connect to the passed database name to be ready to interact with this resource.

import os
import sqlite3
from typing import List

class DatabaseAPI:

    def __init__(self, db_name) -> None:
        os.chdir(os.path.dirname(__file__))
        self._db_name = db_name
        self._connection = self.db_connect()

Database connection handler method

that returns sqlite3 connection if the connection process goes true.

    def db_connect(self) -> sqlite3:
        try:
            return sqlite3.connect(self._db_name)
        except Exception as error:
            raise Exception(f"[-] Error Message:\n{error}\n")

Getting all books

Getting all books from database and Books table by using SELECT * and return list of results.

    def get_all_books(self) -> List[list]:
        cursor = self._connection.cursor()
        response = cursor.execute(f"SELECT * FROM Books;")
        return list(response)

Getting Book by id

like the previous code sample but with adding the WHERE keyword to filter results for the passing book id.

    def get_book_by_id(self, id) -> List:
        cursor = self._connection.cursor()
        response = cursor.execute(f"SELECT * FROM Books WHERE Id = {id};")
        return response

Insert and update a Book

i implement 2 methods that takes a book object/dict and passing it to the INSERT statement then commit this action to the database, another method do the same but it perform update action using SET statement with the book id to update a specific book.

    def insert_book(self, book_data: dict):
        cursor = self._connection.cursor()
        cursor.execute(f"""
        INSERT INTO Books (title, author, description, pages)
	    VALUES ('{book_data['title']}',
			    '{book_data['author']}',
			    '{book_data['description']}',
			    '{book_data['pages']}');""")
        self._connection.commit()

     def update_book(self, id: int, new_data: dict):
        cursor = self._connection.cursor()
        cursor.execute(f"""
        UPDATE Books SET title = '{new_data['title']}',
                         author = '{new_data['author']}',
                         description = '{new_data['description']}',
                         pages = '{new_data['pages']}'
        WHERE Id = {id};""")
        self._connection.commit()

Deleting Book by id

    def delete_book(self, id):
        cursor = self._connection.cursor()
        cursor.execute(f"""DELETE FROM Books WHERE Id = {id};""")
        self._connection.commit()

Put altogether

import os
import sqlite3
from typing import List

class DatabaseAPI:

    def __init__(self, db_name) -> None:
        os.chdir(os.path.dirname(__file__))
        self._db_name = db_name
        self._connection = self.db_connect()

    def db_connect(self) -> sqlite3:
        try:
            return sqlite3.connect(self._db_name)
        except Exception as error:
            raise Exception(f"[-] Error Message:\n{error}\n")

    def get_all_books(self) -> List[list]:
        cursor = self._connection.cursor()
        response = cursor.execute(f"SELECT * FROM Books;")
        return list(response)

    def get_book_by_id(self, id) -> List:
        cursor = self._connection.cursor()
        response = cursor.execute(f"SELECT * FROM Books WHERE Id = {id};")
        return response

    def insert_book(self, book_data: dict):
        cursor = self._connection.cursor()
        cursor.execute(f"""
        INSERT INTO Books (title, author, description, pages)
        VALUES ('{book_data['title']}',
		        '{book_data['author']}',
		        '{book_data['description']}',
		        '{book_data['pages']}');""")
        self._connection.commit()

    def update_book(self, id: int, new_data: dict):
        cursor = self._connection.cursor()
        cursor.execute(f"""
        UPDATE Books SET title = '{new_data['title']}',
                         author = '{new_data['author']}',
                         description = '{new_data['description']}',
                         pages = '{new_data['pages']}'
        WHERE Id = {id};""")
        self._connection.commit()

    def delete_book(self, id):
        cursor = self._connection.cursor()
        cursor.execute(f"""DELETE FROM Books WHERE Id = {id};""")
        self._connection.commit()
       

if __name__ == "__main__":
    databaseObj = DatabaseAPI("booksDB.db")

Building Flask API

Init the flask Application

# Importing Required classes
from flask import Flask, request

# init Flask application
app = Flask(__name__)

# Simple Endpoint to return Hello world message
@app.route("/hello", methods=["GET"])
def home():
    return "Hello World!"

# Running your application with debugging mode on
app.run(debug=True)

using the previous simple lines of code, we have created an endpoint hello which handle GET request HTTP method and return the Hello World message, lets test this API by running our application and test the endpoint using Postman.

Our application is running on Port 5000 in the following URL http://127.0.0.1:5000 lets send Get Request to our application at the following http://127.0.0.1:5000/hello endpoint.

Handling GET requests

Lets create 2 endpoints to handle the two type of GET requests, our user may wanna list all books or get one book by it’s id so ew need to implement this two situations separately.

Get Book by id

getting book from the Books table by it’s passed id parameter after books/ in url

@app.route("/books/<id>", methods=["GET"])
def get_book_by_id(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    result = list(database_api.get_book_by_id(id))
    response = {
        "data":{
            "result": result[0]
        }
    }
    return response

Get all Books

the following code will handle all GET requests for /books end point and return all books that exist in Books table.

@app.route("/books", methods=["GET"])
def get_all_books() -> list:
    database_api = DatabaseAPI("booksDB.db")
    response = {
        "data":{
            "result": database_api.get_all_books()
        }
    }
    return response

Handling POST requests

post requests for adding new book to our database, in this section for POST and PUT methods we need to read the passed data in the request body, we can do that by using request from flask class then access the data key like the following line

request_body = json.loads(request.data.decode())

we load the data in JSON format to be accessible for us as a dict type. after getting needed body parameters we can add new book to the database like the following.

@app.route("/books", methods=["POST"])
def add_new_book() -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    request_body = json.loads(request.data.decode())
    new_book = {
        "title": request_body["title"],
        "author": request_body["author"],
        "description": request_body["description"],
        "pages": request_body["pages"],
    }
    # Add book to database
    database_api.insert_book(new_book)
    return jsonify(new_book)

Handling PUT requests

Returning the updated record after operation done successfully. Note that we pass id after /books/ to apply the update operation to that specific book.

@app.route("/books/<id>", methods=["PUT"])
def update_book(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    request_body = json.loads(request.data.decode())
    new_book = {
        "title": request_body["title"],
        "author": request_body["author"],
        "description": request_body["description"],
        "pages": request_body["pages"],
    }
    # Add book to database
    database_api.update_book(id, new_book)
    return jsonify(new_book)

Handling DELETE requests

Deleting book by it’s id then returning the deleted book record.

@app.route("/books/<id>", methods=["DELETE"])
def delete_book(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    selected_book = list(database_api.get_book_by_id(id))
    response = {
        "data":{
            "result": selected_book[0]
        }
    }
    database_api.delete_book(id)
    return response

Put altogether

at the end our API looks like the following

from flask import Flask, request, jsonify
import json
from cores.database_api import DatabaseAPI

# init Flask application
app = Flask(__name__)

# Simple Endpoint to return Hello world message
@app.route("/hello", methods=["GET"])
def home():
    return "Hello World!"

@app.route("/books", methods=["GET"])
def get_all_books() -> list:
    database_api = DatabaseAPI("booksDB.db")
    response = {
        "data":{
            "result": database_api.get_all_books()
        }
    }
    return response

@app.route("/books/<id>", methods=["GET"])
def get_book_by_id(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    result = list(database_api.get_book_by_id(id))
    response = {
        "data":{
            "result": result[0]
        }
    }
    return response

@app.route("/books", methods=["POST"])
def add_new_book() -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    request_body = json.loads(request.data.decode())
    new_book = {
        "title": request_body["title"],
        "author": request_body["author"],
        "description": request_body["description"],
        "pages": request_body["pages"],
    }
    # Add book to database
    database_api.insert_book(new_book)
    return jsonify(new_book)

@app.route("/books/<id>", methods=["PUT"])
def update_book(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    request_body = json.loads(request.data.decode())
    new_book = {
        "title": request_body["title"],
        "author": request_body["author"],
        "description": request_body["description"],
        "pages": request_body["pages"],
    }
    # Add book to database
    database_api.update_book(id, new_book)
    return jsonify(new_book)

@app.route("/books/<id>", methods=["DELETE"])
def delete_book(id: int) -> str:
    database_api = DatabaseAPI("booksDB.db")
    # Reading request body data in json format
    selected_book = list(database_api.get_book_by_id(id))
    response = {
        "data":{
            "result": selected_book[0]

        }
    }
    database_api.delete_book(id)
    return response

# Running your application with debugging mode on
app.run(debug=True)

Try Your new API

Using the Postman application we will try and test our Implementation.

Test POST Method

• simple test data (raw json data)

{
  "title": "Building Powerful API Using Flask",
  "author": "Hossam hamdy",
  "description": "DESCRIPTION DESCRIPTION DESCRIPTION DESCRIPTION",
  "pages": 100
}

• URL: http://127.0.0.1:5000/books
• Result:

Test GET Method

As we see in the previous test our book is added successfully, now lets try to get this book by id 4

• URL: http://127.0.0.1:5000/books/4
• Result:

Testing PUT Method

Lets update the inserted book to be using NodeJS instead of python.

• simple test data (raw json data)

{
  "title": "Building Powerful API Using NodeJS",
  "author": "hossam hamdy",
  "description": "DESCRIPTION DESCRIPTION DESCRIPTION DESCRIPTION",
  "pages": 100
}

• URL: http://127.0.0.1:5000/books/4
• Result:

Testing DELETE Method

congrats our API is ready and built successfully 😃.

Appendix

Most modern web applications expose APIs that clients can use to interact with the application. A well-designed web API should aim to support:

• Platform independence. Any client should be able to call the API, regardless of how the API is implemented internally. This requires using standard protocols, and having a mechanism whereby the client and the web service can agree on the format of the data to exchange.
• Service evolution. The web API should be able to evolve and add functionality independently from client applications. As the API evolves, existing client applications should continue to function without modification. All functionality should be discoverable so that client applications can fully use it.

You can use this resource for building robust, well-designed APIs Link

Thank You for reading 😃.