FastAPI CRUD Operations with a Database

CRUD stands for Create, Read, Update, and Delete. These four operations cover everything an API does with stored data. This topic shows you how to perform all four using SQLAlchemy sessions inside FastAPI routes.

Setup Assumed from Previous Topics

database.py  → engine, SessionLocal, Base, get_db()
models.py    → User(Base) with id, name, email, is_active
schemas.py   → UserCreate, UserResponse (orm_mode=True)

CREATE — Inserting a New Record

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
import models, schemas

app = FastAPI()

@app.post("/users", response_model=schemas.UserResponse, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Check for duplicate email
    existing = db.query(models.User).filter(models.User.email == user.email).first()
    if existing:
        raise HTTPException(status_code=409, detail="Email already registered")

    db_user = models.User(name=user.name, email=user.email)
    db.add(db_user)     ← stage the new record
    db.commit()         ← write to database
    db.refresh(db_user) ← reload to get generated id
    return db_user

Steps:
  1. Build a SQLAlchemy model instance
  2. db.add()    → tell the session about it
  3. db.commit() → write to disk
  4. db.refresh()→ fetch the auto-generated id

READ — Fetching Records

Get All Users

@app.get("/users", response_model=list[schemas.UserResponse])
def get_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

Get One User by ID

@app.get("/users/{user_id}", response_model=schemas.UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Query chain explained:
  db.query(models.User)               ← SELECT * FROM users
    .filter(models.User.id == user_id)← WHERE id = ?
    .first()                          ← LIMIT 1, returns None if not found

UPDATE — Modifying an Existing Record

@app.put("/users/{user_id}", response_model=schemas.UserResponse)
def update_user(user_id: int, updates: schemas.UserCreate, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    user.name  = updates.name
    user.email = updates.email
    db.commit()
    db.refresh(user)
    return user

Steps:
  1. Fetch existing record (or 404)
  2. Modify attributes directly on the object
  3. db.commit() → saves changes
  4. db.refresh() → reloads updated state

DELETE — Removing a Record

from fastapi import status

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    db.delete(user)
    db.commit()
    return None

Filtering and Searching

# Search by name (case-insensitive)
@app.get("/users/search")
def search_users(name: str, db: Session = Depends(get_db)):
    users = (
        db.query(models.User)
        .filter(models.User.name.ilike(f"%{name}%"))
        .all()
    )
    return users

.ilike("%priya%")  →  finds "Priya", "priya", "PRIYA"
.like("%priya%")   →  case-sensitive match

CRUD at a Glance

Action    Method      Route            DB Operation
──────────────────────────────────────────────────────
Create    POST        /users           db.add(), db.commit()
Read All  GET         /users           db.query().all()
Read One  GET         /users/{id}      db.query().filter().first()
Update    PUT         /users/{id}      modify object, db.commit()
Delete    DELETE      /users/{id}      db.delete(), db.commit()

Key Points

• Use db.add() + db.commit() + db.refresh() to insert a new record.
• Use db.query(Model).filter().first() to fetch one record by a condition.
• Raise HTTPException(404) when a queried record does not exist.
• Modify object attributes directly, then db.commit() to save updates.
• Use db.delete() + db.commit() to remove a record permanently.