FastAPI – SQL Databases
”; Previous Next
In the previous chapter, a Python list 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 compliant 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 like 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 library 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.sqlite import *
SQLALCHEMY_DATABASE_URL = “sqlite:///./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))
publisher = 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
publisher: 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 application 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,
publisher=b1.publisher)
db.add(bk)
db.commit()
db.refresh(bk)
return Books(**b1.dict())
A database session is first established. 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 /list route. When executed, its server response is the list of all records.
@app.get(”/list”, 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.publisher=book.publisher
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”)
Print Page
Previous
Next
Advertisements ”;