Integrating SQLite with FastAPI

In our previous article, we walked through a step-by-step guide to building a simple REST API using FastAPI. Here’s the link..

In this article, we'll take our FastAPI project a step further by replacing the in-memory list with a real database. This will make our API more scalable and persistent.

Note : In previous article we had stored data in list/array.

Which database should we choose?

SQLite : It is well-suited for local applications and testing simple apps. It is file-based, requires no setup, and runs out of the box. There is no need for a username or password, and it is fast for small-scale applications.

A file named test.db will be created in your project directory, acting as the database and storing all the data. If the file is deleted, all the data will be lost.

Why SQLAlchemy?

SQLAlchemy is an ORM (Object Relational Mapper) that helps interact with databases using Python code instead of raw SQL queries.

What changes do we need to make in the code?

First, we need to install the SQLAlchemy package.

pip install fastapi sqlalchemy uvicorn        

If you want to view the data in test.db, install an SQLite GUI. [ link ]

Now, we need to make some changes in main.py

# Import sqlalchemy 

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# Define DB URL

DATABASE_URL = "sqlite:///./test.db"

# Create Engine & Session

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)        

Next, we need to create a database model class.

class ItemDB(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    age = Column(Integer)
    education = Column(String, nullable=True)        

Whenever we add, modify, or delete a student, the API response will return the student's ID.

class ItemResponse(ItemCreate):
    id: int        

The final step is to update the API definitions to add, remove, or modify data using the database's built-in functions. The complete code looks like this:

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import List

# SQLite Database URL
DATABASE_URL = "sqlite:///./test.db"

# Create Engine & Session
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base Class for ORM Models
Base = declarative_base()

# Database Model
class ItemDB(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    age = Column(Integer)
    education = Column(String, nullable=True)

# Create Tables
Base.metadata.create_all(bind=engine)

# Pydantic Model for Request Validation
class ItemCreate(BaseModel):
    name: str
    age: int
    education: str = None

class ItemResponse(ItemCreate):
    id: int

# Dependency for DB Session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# FastAPI App
app = FastAPI()

# Create Item
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = ItemDB(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# Get All Items
@app.get("/items/", response_model=List[ItemResponse])
def get_items(db: Session = Depends(get_db)):
    return db.query(ItemDB).all()

# Get Item by ID
@app.get("/items/{item_id}", response_model=ItemResponse)
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

# Update Item
@app.put("/items/{item_id}", response_model=ItemResponse)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    for key, value in item.dict().items():
        setattr(db_item, key, value)
    
    db.commit()
    db.refresh(db_item)
    return db_item

# Delete Item
@app.delete("/items/{item_id}", response_model=ItemResponse)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    db.delete(db_item)
    db.commit()
    return db_item
        

As mentioned in the previous article, run the code using the command "uvicorn main:app --reload" This will create the test.db file. Play around with the APIs and view the data using an SQLite GUI.

Happy coding!