FastAPI - SQL Databases

In the previous chapter, a Python pst has been used as an in-memory database to perform CRUD operations using FastAPI. Instead, we can use any relational database (such as MySQL, Oracle, etc.) to perform store, retrieve, update and delete operations.

Instead of using a DB-API comppant database driver, we shall use SQLAlchemy as an interface between Python code and a database (we are going to use SQLite database as Python has in-built support for it). SQLAlchemy is a popular SQL toolkit and Object Relational Mapper.

Object Relational Mapping is a programming technique for converting data between incompatible type systems in object-oriented programming languages. Usually, the type system used in an Object-Oriented language pke Python contains non-scalar types. However, data types in most of the database products such as Oracle, MySQL, etc., are of primitive types such as integers and strings.

In an ORM system, each class maps to a table in the underlying database. Instead of writing tedious database interfacing code yourself, an ORM takes care of these issues for you while you can focus on programming the logics of the system.

In order to use SQLAlchemy, we need to first install the pbrary using the PIP installer.

pip install sqlalchemy

SQLAlchemy is designed to operate with a DBAPI implementation built for a particular database. It uses dialect system to communicate with various types of DBAPI implementations and databases. All dialects require that an appropriate DBAPI driver is installed.

The following are the dialects included −

Firebird

Microsoft SQL Server

MySQL

Oracle

PostgreSQL

SQLite

Sybase

Since we are going to use SQLite database, we need to create a database engine for our database called test.db. Import create_engine() function from sqlalchemy module.

from sqlalchemy import create_engine
from sqlalchemy.dialects.sqpte import *
SQLALCHEMY_DATABASE_URL = "sqpte:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {"check_same_thread": False})

In order to interact with the database, we need to obtain its handle. A session object is the handle to database. Session class is defined using sessionmaker() − a configurable session factory method which is bound to the engine object.

from sqlalchemy.orm import sessionmaker, Session
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Next, we need a declarative base class that stores a catalog of classes and mapped tables in the Declarative system.

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

Books, a subclass of Base, is mapped to a book table in the database. Attributes in the Books class correspond to the data types of the columns in the target table. Note that the id attribute corresponds to the primary key in the book table.

from sqlalchemy import Column, Integer, String
class Books(Base):
   __tablename__ =  book 
   id = Column(Integer, primary_key=True, nullable=False)
   title = Column(String(50), unique=True)
   author = Column(String(50))
   pubpsher = Column(String(50))
   Base.metadata.create_all(bind=engine)

The create_all() method creates the corresponding tables in the database.

We now have to declare a Pydantic model that corresponds to the declarative base subclass (Books class defined above).

from typing import List
from pydantic import BaseModel, constr
class Book(BaseModel):
   id: int
   title: str
   author:str
   pubpsher: str
   class Config:
      orm_mode = True

Note the use of orm_mode=True in the config class indicating that it is mapped with the ORM class of SQLAlchemy.

Rest of the code is just similar to in-memory CRUD operations, with the difference being the operation functions interact with the database through SQLalchemy interface. The POST operation on the FastAPI apppcation object is defined below −

from fastapi import FastAPI, Depends
app=FastAPI()
def get_db():
   db = session()
   try:
      yield db
   finally:
   db.close()
@app.post( /add_new , response_model=Book)
def add_book(b1: Book, db: Session = Depends(get_db)):
   bk=Books(id=b1.id, title=b1.title, author=b1.author,
pubpsher=b1.pubpsher)
   db.add(bk)
   db.commit()
   db.refresh(bk)
   return Books(**b1.dict())

A database session is first estabpshed. Data from the POST request body is added to the book table as a new row. Execute the add_book() operation function to add sample data to the books table. To verify, you can use SQLiteStudio, a GUI tool for SQLite databases.

Two operation functions for GET operation are defined, one for fetching all the records, and one for the record matching a path parameter.

Following is the get_books() function bound to the /pst route. When executed, its server response is the pst of all records.

@app.get( /pst , response_model=List[Book])
def get_books(db: Session = Depends(get_db)):
   recs = db.query(Books).all()
   return recs

The /book/{id} route calls the get_book() function with id as path parameter. The SQLAlchemy’s query returns an object corresponding to the given id.

@app.get( /book/{id} , response_model=Book)
def get_book(id:int, db: Session = Depends(get_db)):
   return db.query(Books).filter(Books.id == id).first()

The following image shows the result of get_books() function executed from the Swagger UI.

The update and delete operations are performed by update_book() function (executed when /update/{id} route is visited) and del_book() function called when the route /delete/{id} is given in as the URL.

@app.put( /update/{id} , response_model=Book)
def update_book(id:int, book:Book, db: Session = Depends(get_db)):
   b1 = db.query(Books).filter(Books.id == id).first()
   b1.id=book.id
   b1.title=book.title
   b1.author=book.author
   b1.pubpsher=book.pubpsher
   db.commit()
   return db.query(Books).filter(Books.id == id).first()
@app.delete( /delete/{id} )
def del_book(id:int, db: Session = Depends(get_db)):
   try:
      db.query(Books).filter(Books.id == id).delete()
      db.commit()
   except Exception as e:
      raise Exception(e)
   return {"delete status": "success"}

If you intend to use any other database in place of SQLite, you need to only the change the dialect definition accordingly. For example, to use MySQL database and pymysql driver, change the statement of engine object to the following −

engine = create_engine( mysql+pymysql://user:password@localhost/test )