FastAPI Simple CRUD With MySQL & SQLAlchemy

This article is aimed as a quick reference whenever you guys need to work on FastAPI just like myself. I often spend several months work in certain tech stack such as FastAPI, and then moved to another tech stack and then working back again with FastAPI.

FastAPI provides good documentation on how to install it starting from create virtual environment until running it on local here: https://fastapi.tiangolo.com/#installation

After do installation properly according to the tutorial above we’ll create a simple todo CRUD with MySQL. First let’s create a database named todo in mysql and do import this query command to create table todos

CREATE TABLE IF NOT EXISTS `todos` (
  `id` int NOT NULL AUTO_INCREMENT,
  `title` varchar(150) CHARACTER SET utf8 NOT NULL,
  `status` enum('Doing','Finished') DEFAULT 'Doing',
  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

After you create a database with table above, we need to install sqlalchemy and mysql-connector-python, by this command:

pip install sqlalchemy mysql-connector-python

Our final file structure in root project folder will be as follow:

/
  - main.py
  - database.py
  - models.py
  - schemas.py
  - repositories.py

First we will create database.py as the database connector from our FastAPI application (makesure you replace all capitalize words with proper MySQL connection in your setup):

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

DB_URL = "mysql+mysqlconnector://YOUR_DB_USER:YOUR_DB_PASSWORD@YOUR_DB_URL:YOUR_DB_PORT/YOUR_DB_NAME"

# Initialize database engine by using MySQL
engine = create_engine(DB_URL,echo=True)

# Create a session factory bound to the engine
SessionLocal = sessionmaker(autocommit=False,autoflush=False, bind=engine)

# Create the base class for all ORM models
# The Base object serves as the parent for all database models you define in your application.
Base = declarative_base()

Next step models.py will be created as the model to connect to table we intended to perform CRUD in database.

from pydantic import BaseModel
from sqlalchemy import Column, Integer, ForeignKey, String, DateTime
from database import Base

# We import Base which is created from database.py 
class Todo(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255))
    status = Column(String(255))
    created_at = Column(DateTime)
    updated_at = Column(DateTime)    

we will create schemas.py as the serializer to easily convert python data type into JSON for our API response:

from pydantic import BaseModel
from typing import Optional
from datetime import datetime


class Todo(BaseModel):
    title : str
    status : str


class TodoRead(Todo):
    id: int
    created_at : datetime
    updated_at : datetime


class TodoPatch(BaseModel):
    # These properties are optional to comply with partial update
    # on PATCH request and can be ommitted
    title: Optional[str] = None
    status: Optional[str] = None


class TodoDelete(BaseModel):
    id: int

repositories.py is created to perform any business logic to database:

from sqlalchemy.orm import Session
from datetime import datetime

import models,schemas


def get_todos(db: Session, skip:int=0, limit: int=100):
    return db.query(models.Todo).offset(skip).limit(limit).all()


def get_todo(db: Session, todo_id: int):
    return db.query(models.Todo).filter(models.Todo.id == todo_id).first()


def delete_todo(db: Session, todo_id: int):
    return db.query(models.Todo).filter(models.Todo.id == todo_id).delete()


def update_todo(db: Session, update_todo: models.Todo, todo_id: int):
    todo = get_todo(db, todo_id)
    if todo is not None:        
        todo.title = update_todo.title if update_todo.title != None else todo.title
        todo.status = update_todo.status if update_todo.status != None else todo.status
        todo.updated_at = datetime.now()
        db.commit()
        db.refresh(todo)        
        return todo
    else:
        return None


def create_todo(db: Session, todo:schemas.Todo):
    db_todo = models.Todo(title=todo.title,
                          status=todo.status,
                          created_at=datetime.now(),
                          updated_at=datetime.now())
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo


# NOTE :
# - To perform CRUD need to add object instance to the database session.
# - do commit changes
# - do refresh your instance to contain new data from the database.

Finally, the last piece of our app is main.py as the main entry & routing:

from fastapi import FastAPI, APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from database import Base, SessionLocal
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import Session

import models,schemas, repositories
import logging

#create FastAPI application instance
app = FastAPI()


def get_db():
    db = SessionLocal()
    try : 
        yield db
    finally:
        db.close()


# Create routers for different route groups
todo_router_v1 = APIRouter(prefix="/todos", tags=["todos v1"])

'''
 -  Here we injected schemas / serializer from TodoRead to response since this can 
    show more than one todo result we put use list 
 -  We create a simple repository to perform database process here
 -  Last we put proper http status using status_code decorator.
    Common status codes in the status module:
    status.HTTP_200_OK (default for successful responses)
    status.HTTP_201_CREATED
    status.HTTP_204_NO_CONTENT
    status.HTTP_400_BAD_REQUEST
    status.HTTP_404_NOT_FOUND
    status.HTTP_500_INTERNAL_SERVER_ERROR

'''
@todo_router_v1.get("", summary="List all todos", status_code=status.HTTP_200_OK, response_model=list[schemas.TodoRead])
async def get_todo_v1(db:Session=Depends(get_db)):
    return repositories.get_todos(db)


'''
 -  We use response from TodoRead schema so it'll show all 4 properties of the record.
 -  Incoming request uses Todo schema. According to the schema two parameters are mandatory
    it'll act as a validation as well
'''
@todo_router_v1.post("", summary="Create a new todo", status_code=status.HTTP_201_CREATED, response_model=schemas.TodoRead)
async def post_todo_v1(todo:schemas.Todo, db:Session=Depends(get_db)):
    return repositories.create_todo(db, todo)


'''
 -  To get todo detail we only need one parameter to fetch a todo based on its id
 -  Throw http exception if record is not found
'''
@todo_router_v1.get("/{todo_id}", summary="Get todo detail", status_code=status.HTTP_200_OK, response_model=schemas.TodoRead)
async def patch_todo_v1(todo_id: str, db:Session=Depends(get_db)):
    todo = repositories.get_todo(db, todo_id)
    if todo is not None:
        return todo
    else:
        raise HTTPException(status_code=400, detail="Todo is not found")


'''
 -  Patch is only partial update therefore we use TodoPatch schemas which we put optional parameter
    for title & status. If any of these parameters are added then the value of these parameters will
    be updated
 '''
@todo_router_v1.patch("/{todo_id}", summary="Partial update todo", status_code=status.HTTP_200_OK, response_model=schemas.TodoRead)
async def patch_todo_v1(todo_id: str, update_todo:schemas.TodoPatch, db:Session=Depends(get_db)):
    todo = repositories.get_todo(db, todo_id)
    if todo is not None:
        repositories.update_todo(db, update_todo, todo_id)
        db.commit()
        db.refresh(todo)      

        return todo
    else:
        raise HTTPException(status_code=400, detail="Todo is not found")


'''
 -  Put request means all update according to the REST best practice. On this case 
    the required parameters are identical with post so we use the same schema for 
    incoming request with post which is Todo
'''
@todo_router_v1.put("/{todo_id}", summary="Update todo",status_code=status.HTTP_200_OK, response_model=schemas.TodoRead)
async def put_todo_v1(todo_id: str, update_todo:schemas.Todo, db:Session=Depends(get_db)):
    todo = repositories.get_todo(db, todo_id)
    if todo is not None:
        repositories.update_todo(db, update_todo, todo_id)
        db.commit()
        db.refresh(todo)      

        return todo
    else:
        raise HTTPException(status_code=400, detail="Todo is not found")


@todo_router_v1.delete("/{todo_id}",status_code=status.HTTP_204_NO_CONTENT, summary="Delete todo")
async def delete_todo_v1(todo_id: str, db:Session=Depends(get_db)):
    todo = repositories.get_todo(db, todo_id)
    if todo is not None:
        db.delete(todo)
        db.commit()
        return
    else:
        raise HTTPException(status_code=400, detail="Todo is not found")


# Create versioning for API endpoint by incorporating todo_router_v1 into FastAPI application instance
app.include_router(todo_router_v1, prefix="/v1")

We follow best naming api convention practice by:

• Adding versioning (in this case is v1) followed by resource name.
• Resource name is plural.
• Resource are representated by noun and URIs shouldn’t indicating CRUD operations.

Therefore, if you run the app locally by this command:

fastapi def main.py

to show FastAPI docs for example http://localhost:8000/docs our API will look like this :

Now we have implemented simple CRUD from FastAPI and how to implement repository pattern. Repository for this article can be accessed here: https://github.com/gerry-sabar/fastapi-todo