Hello, there!

If you are new to FastAPI, Please read my previous post first.

In this article you'll learn how to build a CRUD application using FastAPI, SQLAlchemy and MySQL database.

Objectives:

• Define SQLAlchemy models.
• Build CRUD operations to interact with the database.
• Define Pydantic models for data validation and conversion .
• Build REST APIs with FastAPI.

You can find the code at my GitHub Repository . Leave a star if you find it useful.

Connecting the database:

Let's configure the SQLAlchemy so that it knows how to establish a connection with the database.

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+mysqldb://user:password@host/db_name"

db_engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

Base = declarative_base()

What is SQLAlchemy?

SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL.

It provides a full suite of well known enterprise-level persistence patterns, designed for efficient and high-performing database access, adapted into a simple and Pythonic domain language.

Read more about SQLAlchemy on their official site.

Defining Data Models using SQLAlchemy:

Our application has only one model. With the help of SQLAlchemy we can define our model to interact with our table "car" like this:

# models.py

from sqlalchemy.schema import Column
from sqlalchemy.types import String, Integer, Enum
from database import Base
import enum


class FuelType(str, enum.Enum):
    petrol = "Petrol"
    diesel = "Diesel"


class CarInfo(Base):
    __tablename__ = "car"

    id = Column(Integer, primary_key=True, index=True)
    manufacturer = Column(String)
    modelName = Column(String)
    cc = Column(Integer)
    onRoadPrice = Column(Integer)
    seatingCapacity = Column(Integer)
    gearBox = Column(Integer)
    fuelType = Column(Enum(FuelType))

The structure of the MySQL table will be like this

Create table statement

create table car (
id int AUTO_INCREMENT PRIMARY KEY, 
manufacturer varchar(30) NOT NULL, 
modelName varchar(30) NOT NULL,
cc int NOT NULL, 
onRoadPrice int NOT NULL, 
seatingCapacity int NOT NULL, 
gearBox int NOT NULL, 
fuelType enum ('Petrol', 'Diesel') NOT NULL
);

Data Validation and Conversion:

Before getting into the API implementation, Let's define the pydantic models for the incoming and outgoing data validation and conversion of request handlers / API endpoints.

# schemas.py
from pydantic import BaseModel
from models import FuelType
from typing import Optional, List


# TO support creation and update APIs
class CreateAndUpdateCar(BaseModel):
    manufacturer: str
    modelName: str
    cc: int
    onRoadPrice: int
    seatingCapacity: int
    gearBox: int
    fuelType: FuelType


# TO support list and get APIs
class Car(CreateAndUpdateCar):
    id: int

    class Config:
        orm_mode = True


# To support list cars API
class PaginatedCarInfo(BaseModel):
    limit: int
    offset: int
    data: List[Car]

Now we are done with the schema definition for all data exchanges.

CRUD Operations and Custom Exceptions:

Let's define the helper functions to perform actual CRUD operations on the db. These functions will be used inside the REST API endpoints.

# crud.py
from typing import List
from sqlalchemy.orm import Session
from exceptions import CarInfoInfoAlreadyExistError, CarInfoNotFoundError
from models import CarInfo
from schemas import CreateAndUpdateCar


# Function to get list of car info
def get_all_cars(session: Session, limit: int, offset: int) -> List[CarInfo]:
    return session.query(CarInfo).offset(offset).limit(limit).all()


# Function to  get info of a particular car
def get_car_info_by_id(session: Session, _id: int) -> CarInfo:
    car_info = session.query(CarInfo).get(_id)

    if car_info is None:
        raise CarInfoNotFoundError

    return car_info


# Function to add a new car info to the database
def create_car(session: Session, car_info: CreateAndUpdateCar) -> CarInfo:
    car_details = session.query(CarInfo).filter(CarInfo.manufacturer == car_info.manufacturer, CarInfo.modelName == car_info.modelName).first()
    if car_details is not None:
        raise CarInfoInfoAlreadyExistError

    new_car_info = CarInfo(**car_info.dict())
    session.add(new_car_info)
    session.commit()
    session.refresh(new_car_info)
    return new_car_info


# Function to update details of the car
def update_car_info(session: Session, _id: int, info_update: CreateAndUpdateCar) -> CarInfo:
    car_info = get_car_info_by_id(session, _id)

    if car_info is None:
        raise CarInfoNotFoundError

    car_info.manufacturer = info_update.manufacturer
    car_info.modelName = info_update.modelName
    car_info.fuelType = info_update.fuelType
    car_info.cc = info_update.cc
    car_info.gearBox = info_update.gearBox
    car_info.onRoadPrice = info_update.onRoadPrice
    car_info.seatingCapacity = info_update.seatingCapacity

    session.commit()
    session.refresh(car_info)

    return car_info


# Function to delete a car info from the db
def delete_car_info(session: Session, _id: int):
    car_info = get_car_info_by_id(session, _id)

    if car_info is None:
        raise CarInfoNotFoundError

    session.delete(car_info)
    session.commit()

    return

Define custom exception to send proper exception response.

# exceptions.py
class CarInfoException(Exception):
    ...


class CarInfoNotFoundError(CarInfoException):
    def __init__(self):
        self.status_code = 404
        self.detail = "Car Info Not Found"


class CarInfoInfoAlreadyExistError(CarInfoException):
    def __init__(self):
        self.status_code = 409
        self.detail = "Car Info Already Exists"

REST API Implementation:

Define the API endpoints like this

# api.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi_utils.cbv import cbv
from sqlalchemy.orm import Session
from crud import get_all_cars, create_car, get_car_info_by_id, update_car_info, delete_car_info
from database import get_db
from exceptions import CarInfoException
from schemas import Car, CreateAndUpdateCar, PaginatedCarInfo

router = APIRouter()


# Example of Class based view
@cbv(router)
class Cars:
    session: Session = Depends(get_db)

    # API to get the list of car info
    @router.get("/cars", response_model=PaginatedCarInfo)
    def list_cars(self, limit: int = 10, offset: int = 0):

        cars_list = get_all_cars(self.session, limit, offset)
        response = {"limit": limit, "offset": offset, "data": cars_list}

        return response

    # API endpoint to add a car info to the database
    @router.post("/cars")
    def add_car(self, car_info: CreateAndUpdateCar):

        try:
            car_info = create_car(self.session, car_info)
            return car_info
        except CarInfoException as cie:
            raise HTTPException(**cie.__dict__)


# API endpoint to get info of a particular car
@router.get("/cars/{car_id}", response_model=Car)
def get_car_info(car_id: int, session: Session = Depends(get_db)):

    try:
        car_info = get_car_info_by_id(session, car_id)
        return car_info
    except CarInfoException as cie:
        raise HTTPException(**cie.__dict__)


# API to update a existing car info
@router.put("/cars/{car_id}", response_model=Car)
def update_car(car_id: int, new_info: CreateAndUpdateCar, session: Session = Depends(get_db)):

    try:
        car_info = update_car_info(session, car_id, new_info)
        return car_info
    except CarInfoException as cie:
        raise HTTPException(**cie.__dict__)


# API to delete a car info from the data base
@router.delete("/cars/{car_id}")
def delete_car(car_id: int, session: Session = Depends(get_db)):

    try:
        return delete_car_info(session, car_id)
    except CarInfoException as cie:
        raise HTTPException(**cie.__dict__)

Include the API routes into FastAPI app

# main.py
# Import FastAPI
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
import api

# Initialize the app
app = FastAPI()

app.include_router(api.router)


# GET operation at route '/'
@app.get('/')
def root_api():
    return {"message": "Welcome to Balasundar's Technical Blog"}

Let's start the FastAPI server! Open a terminal window inside the project directory and run the following command.

uvicorn --reload main:app

The output of the command should be like this

FastAPI generates automatic API documentation using Swagger. You can access the ReDoc by visiting localhost:8000/redoc and Swagger documentation at 127.0.0.1:8000/docs

Now you can test these APIs using any of the API clients tools like Postman or you can use Swagger.

Hooray!!. We are done implementing a very simple CRUD application using FastAPI, SQLAlchemy and MySQL.

You can download the entire project from my GitHub Repository . Leave a star if you find it useful.

Hope it helped.

Thank you 🙏 for reading.