Building CRUD application with FastAPI

In this article we will create a simple Todo application based on FastAPI framework to demonstrate CRUD operation. If you are new to FastAPI framework I would highly suggest you go through Getting Started with FastAPI.

While working on this Todo application you will get to know about the following things.

Extend the FastAPI application from Getting Started with FastAPI to implement Todo app functionality.
Database connectivity and creating Database Tables
How REST APIs work
How to use pydantic models in the APIs
Use of FastAPI swagger Documentation

In this Todo Applications tutorial, User would be able to perform

Create Todo item
Read Todo item/items
Update Todo item
Delete Todo item

Let’s Begin!

Setup

Before writing actual code we need to set few things first

Create a project folder, we are going to call it todo_application ( You can choose name of your choice, no restrictions 😛 )
Create a virtual environment in Python, If you have no idea about virtual environment then follow Setting up Python Environment article.
Setup basic FastAPI application boilerplate code by following Getting Started with FastAPI

Till now your code looks something like this

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}
Code language: Python (python)

Let’s add the endpoints which will handle all the CRUD operations, after adding your code would look something like this

from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return "Todo Application"

@app.post("/todo")
def create_todo():
    return "Create a Todo item"

@app.get("/todo/{id}")
def read_todo(id: int):
    return "Read Todo item using Todo unique {id}"

@app.put("/todo/{id}")
def update_todo(id: int):
    return "Update Todo item using Todo unique {id}"

@app.delete("/todo/{id}")
def delete_todo(id: int):
    return "Delete a Todo using unique {id}"

@app.get("/todo")
def read_all_todos():
    return "Read all the created Todo's"
Code language: Python (python)

Here you can see that we have used different types of request for each endpoints, such as

@app.post – Post Endpoint is used to Create records.
@app.get – This is a GET endpoint, which is used to fetch/Read data from the server , It can be used to fetch single record using some sort of identification ( In our case we are using ‘id’ ) or we can fetch multiple records based on some conditions ( such as Todo’s created in a selected date range ) and we can also fetch entire records as we are doing in the last request.
@app.put – PUT endpoint is used to Update previously created record using some sort of identification such as id of the Todo record in our case.
@app.delete – Delete Endpoint as the name suggests is used to Delete the created Record.

As you can see all the endpoints are mainly focusing on C.R.U.D, Which is the main goal of this article 🙂

Connecting the Database

Create a python file as database.py next to your main.py file and write the following code in it.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Create a sqlite engine instance
engine = create_engine("sqlite:///todo.db")

# Create a DeclarativeMeta instance
Base = declarative_base()

# Create SessionLocal class from sessionmaker factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
Code language: Python (python)

Now open up your terminal, activate your environment and install Sqlalchemy using pip with the following command

(pythonwarriors) G:\Todo>pip install sqlalchemy
Collecting sqlalchemy
  Downloading https://files.pythonhosted.org/packages/55/74/a3f79e7ed1af6f669f4e630cecc8d672b201fe8cfe5a39c8bd47f4209bf7/SQLAlchemy-1.4.29-cp37-cp37m-win_amd64.whl (1.5MB)
     |████████████████████████████████| 1.6MB 139kB/s
Collecting importlib-metadata; python_version < "3.8" (from sqlalchemy)
  Downloading https://files.pythonhosted.org/packages/f9/6c/a14560ec00a14f50fdb3e91665563500b55f3c672e621c3ef159d351e9f7/importlib_metadata-4.10.0-py3-none-any.whl
Collecting greenlet!=0.4.17; python_version >= "3" and (platform_machine == "aarch64" or (platform_machine == "ppc64le" or (platform_machine == "x86_64" or (platform_machine == "amd64" or (platform_machine == "AMD64" or (platform_machine == "win32" or platform_machine == "WIN32")))))) (from sqlalchemy)
  Downloading https://files.pythonhosted.org/packages/f5/34/adc2134c9567dd99254f20e6981a9006a5767dfed287eb94f273ec51e092/greenlet-1.1.2-cp37-cp37m-win_amd64.whl (101kB)
     |████████████████████████████████| 102kB 598kB/s
Collecting zipp>=0.5 (from importlib-metadata; python_version < "3.8"->sqlalchemy)
  Downloading https://files.pythonhosted.org/packages/52/c5/df7953fe6065185af5956265e3b16f13c2826c2b1ba23d43154f3af453bc/zipp-3.7.0-py3-none-any.whl
Requirement already satisfied: typing-extensions>=3.6.4; python_version < "3.8" in c:\users\jsb\miniconda3\envs\pythonwarriors\lib\site-packages (from importlib-metadata; python_version < "3.8"->sqlalchemy) (4.0.1)
Installing collected packages: zipp, importlib-metadata, greenlet, sqlalchemy
Successfully installed greenlet-1.1.2 importlib-metadata-4.10.0 sqlalchemy-1.4.29 zipp-3.7.0
Code language: Python (python)

Sqlalchemy – We are going to use sqlchemy python library to interact with our database from the fastAPI application.
SQLite – It’s a lightweight database library that we are going to use in our application

Create Tables

Create a file and name it as models.py in the same directory as main.py and write the following code

from sqlalchemy import Column, Integer, String
from database import Base

class ToDo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(256))
Code language: Python (python)

This will create the necessary table into our database. All the CRUD operations are going to be performed on this table. Now we are all set just one thing is remainig which is to use pydantic models toc reate request and response model objects so that each of our endpoint knows about the format in which data is to be recieved and sent which makes it easier to debug if there is any fault in the data.
So to do this create a file and name it as schema.py in the same root directory and write the following code in it.

from pydantic import BaseModel

# Create ToDo Schema (Pydantic Model)
class ToDoCreate(BaseModel):
    task: str

# Complete ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
    id: int
    task: str

    class Config:
        orm_mode = True
Code language: Python (python)

Now after importing all these new files in our main.py , It would look something like this

from typing import List
from fastapi import FastAPI, status, HTTPException, Depends
from database import Base, engine, SessionLocal
from sqlalchemy.orm import Session
import models
import schemas

# Create the database
Base.metadata.create_all(engine)

# Initialize app
app = FastAPI()

# provides database session to each request upon calling
def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
Code language: Python (python)

Here the get_session() method will be used in all of our endpoints to provide interaction with the database and perform operations.

CRUD Operations

Before moving to be working on the API endpoints let’s take a look at the swagger Documentation of our application so far. To do that you need to open up your terminal and activate your python environment and inside your Todo project directory terminal type the following uvicorn main:app --reload . It will launch our application on 8000 port , to view swagger documentation go to 127.0.0.1:8000/docs and it should look like this

And here is the updated code for POST request that we will be using to make request to create new task for our Todo application.

@app.post("/todo", response_model=schemas.ToDoAll, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate, session: Session = Depends(get_session)):

    # create an instance of Todo Model
    todo_obj = models.ToDo(task = todo.task)

    # Add the object into our database Table
    session.add(todo_obj)
    session.commit()
    session.refresh(todo_obj)

    # return the todo object
    return todo_obj
Code language: PHP (php)

Replace the previous code with the above code and hit ctrl + s . It’ll automatically relaod your server with the new code. Now go back again to the documentation and check the POST method. It should look something like this

Testing the Code

Click on try it out for the post request and add a new task for example: “First task” in the box as

{
"task": "First task"
}
Code language: Python (python)

And click on execute button. If all goes correct you would get a response in this format with 201 as status code.

{
  "id": 1,
  "task": "First task"
}
Code language: Python (python)

It means our POST API is working correctly. Here is the code for all the endpoints, Replace all the previous sample code with the given below code

@app.get("/")
def root():
    return "Welcome to Todo Application. Built with FastAPI and ❤️. \n Please share https://pythonwarriors.com with your friends"

@app.post("/todo", response_model=schemas.ToDoAll, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate, session: Session = Depends(get_session)):

    # create an instance of Todo Model
    todo_obj = models.ToDo(task = todo.task)

    # Add the object into our database Table
    session.add(todo_obj)
    session.commit()
    session.refresh(todo_obj)

    # return the todo object
    return todo_obj

@app.get("/todo/{id}", response_model=schemas.ToDoAll)
def read_todo(id: int, session: Session = Depends(get_session)):

    # Fetch todo record using id from the table
    todo_obj = session.query(models.ToDo).get(id)

    # Check if there is record with the provided id, if not then Raise 404 Exception
    if not todo_obj:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo_obj

@app.put("/todo/{id}", response_model=schemas.ToDoAll)
def update_todo(id: int, task: str, session: Session = Depends(get_session)):

    # Fetch todo record using id from the table
    todo_obj = session.query(models.ToDo).get(id)

    # If the record is present in our DB table then update 
    if todo_obj:
        todo_obj.task = task
        session.commit()

    # if todo item with given id does not exists, raise exception and return 404 not found response
    if not todo_obj:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return todo_obj

@app.delete("/todo/{id}", response_model = str)
def delete_todo(id: int, session: Session = Depends(get_session)):

    # Fetch todo record using id from the table
    todo_obj = session.query(models.ToDo).get(id)

    # Check if Todo record is present in our Database,If not then raise 404 error
    if todo_obj:
        session.delete(todo_obj)
        session.commit()
    else:
        raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")

    return f"Todo task with id {id} successfully deleted"

@app.get("/todo", response_model = List[schemas.ToDoAll])
def read_todo_list(session: Session = Depends(get_session)):

    # get all todo items
    todo_list = session.query(models.ToDo).all()

    return todo_list

Code language: Python (python)

Once again save the file and go check the swagger documentation and play around with it. Create new Todos, Read them, Update them and Delete few as that’s what CRUD actually means :P.

You can get the full code from github by clicking here FastAPI Todo Application

Setup

Connecting the Database

Create Tables

CRUD Operations

Testing the Code

You Might Also Like

Exploring the Ethical Implications of GPT-3: The Pros, Cons, and Potential Risks of the World’s Most Powerful Language Generation Model

A Comprehensive Comparison of GPT-3, BERT, and Transformer-XL

Getting Started with FastAPI

Leave a Reply Cancel reply