Creating REST API using FastAPI

• Install FastAPI
• Hello World REST API
• Employees API
• Consuming Employees API

Installing FastAPI

pip install "fastapi[all]"

Hello World REST API

hello.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
async def hello():
    return "Hello World"

>uvicorn hello:app --reload
←[32mINFO←[0m:     Will watch for changes in these directories: ['c:\\dev\\python\\fastapidemo']
←[32mINFO←[0m:     Uvicorn running on ←[1mhttp://127.0.0.1:8000←[0m (Press CTRL+C to quit)
←[32mINFO←[0m:     Started reloader process [←[36m←[1m27104←[0m] using ←[36m←[1mwatchgod←[0m
←[32mINFO←[0m:     Started server process [←[36m8520←[0m]
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Application startup complete.

Employee API

CREATE TABLE EMPLOYEES (
    Id       INTEGER      PRIMARY KEY AUTOINCREMENT,
    fullname VARCHAR (30) NOT NULL,
    job      VARCHAR (10),
    salary   INTEGER (10)
);

insert into employees(fullname,job,salary) values('Andy Roberts','SP',1500000)

employee.py

import sqlite3
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

DBPATH = r"c:\dev\python\fastapidemo\hr.db"

# Model to represent employee 
class Employee(BaseModel): 
    name: str
    job: str
    salary: int

# Convert tuple containing employee details to dict 
def record_to_dict(emp):
    return {'id': emp[0], 'name': emp[1], 'job': emp[2], 'salary': emp[3]}


app = FastAPI()

# Handle GET request 
@app.get("/employees")
async def get_all_employees():
    with sqlite3.connect(DBPATH) as con:
        cur = con.cursor()
        cur.execute("select * from employees order by salary desc")  
        employees = []

        for emp in cur.fetchall():
            employees.append(record_to_dict(emp))  # Convert tuple to dict

        cur.close()
        return employees


#Handle POST request to insert row into table 
@app.post("/employees")
async def add_employee(value: Employee):
    with sqlite3.connect(DBPATH) as con:
        cur = con.cursor()
        try:
            cur.execute("insert into employees(fullname,job,salary) values(?,?,?)",
                        (value.name, value.job, value.salary))
            con.commit()  # make insertion permanent
            return value
        except sqlite3.DatabaseError as ex:
            print("Insertion Error : ", ex)
            raise HTTPException(status_code=500, detail=str(ex))

        cur.close()
        con.close()


# Handle DELETE request to delete row 
@app.delete("/employees/{id}")
async def delete_employee(id: int):
    with sqlite3.connect(DBPATH) as con:
        try:
            cur = con.cursor()
            cur.execute("delete from employees where id = ?", (id,))
            if cur.rowcount == 1:
                con.commit()    # make delete permanent
                return None
            else:
                raise HTTPException(status_code=404, detail="Id not found")
        except sqlite3.DatabaseError as ex:
            print("Deletion Error : ", ex)
            raise HTTPException(status_code=500, detail=str(ex))

>uvicorn employee:app --reload

Consuming Employee API

Making GET Request

Making POST Request

Making DELETE Request