Hello in this tutorial, we will implement the CRUD operations using the FastAPI framework in Python. For the database, we will be using SQLite.
1. Introduction
FastAPI is a fast and high-performance web framework for building api with python 3.6+. It offers many features like:
- High performance
- Automatic interactive code generation
- Offers great editor support and documentation
1.1 SQLite
SQLite is a software library that provides a relational database management system. It is lightweight in terms of setup, database administration, and required resources. It is self-contained, serverless, zero-configuration, transactional.
- Self-contained means that it require minimal support from the operating system or any external library
- Zero-configuration means that no external installation is required before using it
- Transactional means it is fully ACID-compliant i.e. all queries and changes are atomic, consistent, isolated, and durable
1.2 Setting up Python
If someone needs to go through the Python installation on Windows, please watch this link. You can download the Python from this link.
2. CRUD Operations using Python FastAPI
I am using JetBrains PyCharm as my preferred IDE. You’re free to choose the IDE of your choice. Fig. 1 represents the project structure for this tutorial.
The file named – movie_database.db will be generated dynamically during the application run.
2.1 Creating a requirements file
Add the below code to the requirements file. The file will be responsible to download and install the packages required for this tutorial.
requirements.txt
fastapi sqlalchemy pydantic
2.2 Creating the database config
Create the database configuration file. The file will be responsible for handling the SQLite database and interacting with the movie database.
db_handler.py
# Handling database
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./movie_database.db"
# creating engine
# Setting check_same_thread to False so that the returned connection may be shared across multiple threads
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
# bind – An optional Connectable, will assign the bind attribute on the MetaData instance
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# A simple constructor that allows initialization
Base = declarative_base()
2.3 Creating the model class
Create the model class responsible for creating the table structure.
model.py
# Model class
from sqlalchemy import Boolean, Column, Integer, String
from db_handler import Base
class Movies(Base):
# Setting constraints on the table structure
__tablename__ = "movie"
id = Column(Integer, primary_key=True, autoincrement=True, index=True, nullable=False)
movie_id = Column(String)
movie_name = Column(String(255))
director = Column(String(100))
geners = Column(String)
membership_required = Column(Boolean)
cast = Column(String(255))
streaming_platform = Column(String)
2.4 Creating the schema class
Create the schema class responsible for creating the schema and use it during the crud operations.
schema.py
# Schema class
from typing import Optional
from pydantic import BaseModel
class MovieBase(BaseModel):
movie_name: str
director: str
geners: str
cast: str
class MovieAdd(MovieBase):
movie_id: int
# Optional[str] is just a shorthand or alias for Union[str, None].
streaming_platform: Optional[str] = None
membership_required: bool
# Behaviour of pydantic can be controlled via the Config class on a model
# to support models that map to ORM objects. Config property orm_mode must be set to True
class Config:
orm_mode = True
class Movie(MovieAdd):
id: int
# Behaviour of pydantic can be controlled via the Config class on a model
# to support models that map to ORM objects. Config property orm_mode must be set to True
class Config:
orm_mode = True
class UpdateMovie(BaseModel):
# Optional[str] is just a shorthand or alias for Union[str, None].
streaming_platform: Optional[str] = None
membership_required: bool
# Behaviour of pydantic can be controlled via the Config class on a model
# to support models that map to ORM objects. Config property orm_mode must be set to True
class Config:
orm_mode = True
2.5 Creating the crud class
Create the crud class responsible for performing the sql operations.
crud.py
# Crud operation methods
from sqlalchemy.orm import Session
import model
import schema
def get_all(db: Session):
print("fetching records")
return db.query(model.Movies).all()
def get_by_id(db: Session, sl_id: int):
print("fetching record id={}".format(sl_id))
return db.query(model.Movies).filter(model.Movies.id == sl_id).first()
def add(db: Session, movie: schema.MovieAdd):
mv_details = model.Movies(
movie_id=movie.movie_id,
movie_name=movie.movie_name,
director=movie.director,
geners=movie.geners,
membership_required=movie.membership_required,
cast=movie.cast,
streaming_platform=movie.streaming_platform)
print("saving new record")
db.add(mv_details)
db.commit()
db.refresh(mv_details)
return model.Movies(**movie.dict())
def delete(db: Session, sl_id: int):
try:
print("deleting record id={}".format(sl_id))
db.query(model.Movies).filter(model.Movies.id == sl_id).delete()
db.commit()
except Exception as e:
raise Exception(e)
def update(db: Session, sl_id: int, details: schema.UpdateMovie):
print("updating record id={}".format(sl_id))
db.query(model.Movies).filter(model.Movies.id == sl_id).update(vars(details))
db.commit()
return db.query(model.Movies).filter(model.Movies.id == sl_id).first()
2.6 Creating the application class
Create the application controller responsible for handling the incoming requests from the user and interact with the database to show the results.
application.py
# Main class
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
import crud
import model
import schema
from db_handler import SessionLocal, engine
model.Base.metadata.create_all(bind=engine)
# Initializing the app
app = FastAPI(title="CRUD Operations using Python FastAPI")
# Database dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Get all movies
@app.get('/get-all', response_model=List[schema.Movie])
def get_all(db: Session = Depends(get_db)):
return crud.get_all(db=db)
# Save new movie
@app.post('/add', response_model=schema.MovieAdd)
def add(movie: schema.MovieAdd, db: Session = Depends(get_db)):
movie_id = crud.get_by_id(db=db, sl_id=movie.movie_id)
if movie_id:
print("Resource conflict")
raise HTTPException(status_code=409, detail=f"Resource id {movie_id} already exist")
return crud.add(db=db, movie=movie)
# Delete a movie by id
@app.delete('/delete')
def delete(sl_id: int, db: Session = Depends(get_db)):
details = crud.get_by_id(db=db, sl_id=sl_id)
if not details:
print("Entity not found")
raise HTTPException(status_code=404, detail=f"Resource not found")
try:
crud.delete(db=db, sl_id=sl_id)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Unable to delete: {e}")
return {"status": "accepted", "code": "202", "message": "Resource deleted"}
# Update a movie by id
@app.put('/update', response_model=schema.Movie)
def update(sl_id: int, update_param: schema.UpdateMovie, db: Session = Depends(get_db)):
details = crud.get_by_id(db=db, sl_id=sl_id)
if not details:
print("Entity not found")
raise HTTPException(status_code=404, detail=f"Resource not found")
return crud.update(db=db, details=update_param, sl_id=sl_id)
# Driver code
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="localhost", port=7001, log_level="debug")
3. Run the application
Run the application.py python script once the code is completed and if everything goes well the application will be started on the port number – 7001 as shown in the below logs.
Application logs
INFO: Started server process [44004] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://localhost:7001 (Press CTRL+C to quit)
4. Demo
Open up the browser of your choice and hit the swagger documentation endpoint generated via the FastAPI. The documentation will list the endpoints created above.
Documentation endpoint
http://localhost:7001/docs
If everything goes well the documentation page will be shown as in Fig. 1 and you can use the endpoints to set up a playground for the application.
That is all for this tutorial and I hope the article served you with whatever you were looking for. Happy Learning and do not forget to share!
5. Summary
In this tutorial, we learned about the SQLite sqlalchemy crud operation and implemented the FastAPI framework to quickly set up our application. You can download the source code of this tutorial from the Downloads section.
6. Download the Project
This was a tutorial on how to implement CRUD operations using the FastAPI framework.
You can download the full source code of this example here: CRUD Operations using Python FastAPI
