SQLAlchemy CRUD Operations

You’ve mastered models and sessions — now it’s time to actually do something with them. SQLAlchemy only shines when you can move data from your Python objects into a database and back — that’s CRUD: Create, Read, Update, Delete — the four operations every persistent-backed API relies on. Without a fluent, repeatable CRUD pattern, you’ll write the same queries over and over, and your FastAPI endpoints will become a tangled mess of session management and boilerplate. In this lesson, you’ll learn a clean, reproducible CRUD pattern with SQLAlchemy that you’ll use in every project from here on out.

The problem this lesson solves

Imagine you’ve just defined a User model with a session and an engine. Now what? You need to write code to insert a new user, fetch a user by ID, list all users, update an email, delete an inactive account. The naive approach — inline query after query in your route handlers — leads to:

• Duplicate session and commit logic across every endpoint
• Inconsistent error handling — some handlers roll back, others don’t
• Hard-to-test code — queries are tangled with HTTP logic

Performing CRUD operations with SQLAlchemy solves this by giving you a centralized, session-based pattern that is safe, predictable, and reusable. With just a few functions you can cover all your data persistence needs, and your FastAPI endpoints become thin wrappers around a rock-solid data layer.

Core concept / mental model

Think of CRUD as a conversation with a database. Your session is the translator, your model is the vocabulary, and the query is the question you ask.

• Create — you insert a new record into the database (INSERT)
• Read — you retrieve one or more records (SELECT)
• Update — you modify an existing record (UPDATE)
• Delete — you remove a record (DELETE)

The session as a workbench

A SQLAlchemy Session is your transactional workbench. You open the session, perform operations, and commit the changes. If anything goes wrong, you roll back to a clean state. The session also gives you a unit of work — meaning all your changes are collected and sent to the database in one batch when you commit.

Key insight: The session is not the database connection — it’s an ORM-level wrapper that manages connections, transactions, and identity. You can have many sessions over the same database connection pool.

Query object as a builder

session.query(Model) returns a Query object that you chain filters, order, and limits onto. It’s lazy — it doesn’t hit the database until you actually iterate, call .all(), .first(), or .scalar(). This model underpins all your read operations.

How it works step by step

Here’s the universal CRUD blueprint using SQLAlchemy’s session-based API:

1. Create a session — from your SessionLocal factory.
2. Create — instantiate a model object, add it with session.add(), commit.
3. Read — use session.query() or session.get(Model, id) to fetch.
4. Update — fetch the object, modify attributes, commit.
5. Delete — fetch, call session.delete(), commit.
6. Close — ensure you close the session (or let FastAPI’s dependency handle it).

The standard CRUD function set

# crud.py
from sqlalchemy.orm import Session
from . import models, schemas

def get_user(db: Session, user_id: int):
    """Read a single user by ID."""
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    """Read a user by email."""
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    """Read a list of users with pagination."""
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    """Create a new user."""
    db_user = models.User(email=user.email, hashed_password=user.hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, updates: dict):
    """Update user fields with a dict of attributes."""
    db_user = get_user(db, user_id)
    if db_user:
        for key, value in updates.items():
            setattr(db_user, key, value)
        db.commit()
        db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    """Delete a user by ID, return True if deleted."""
    db_user = get_user(db, user_id)
    if db_user:
        db.delete(db_user)
        db.commit()
        return True
    return False

Hands-on walkthrough

Let’s apply this in a FastAPI route layer. We’ll assume you already have database.py with SessionLocal and a get_db dependency.

File 1: database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

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

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

# Dependency to get a DB session per request
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

File 2: models.py

from sqlalchemy import Column, Integer, String
from .database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

File 3: schemas.py (Pydantic)

from pydantic import BaseModel

class UserCreate(BaseModel):
    email: str
    hashed_password: str

class UserUpdate(BaseModel):
    email: str | None = None
    hashed_password: str | None = None

class User(BaseModel):
    id: int
    email: str
    hashed_password: str
    class Config:
        orm_mode = True

File 4: main.py — Wire CRUD to endpoints

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import Base, engine, get_db

Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.post("/users/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Check email uniqueness
    existing = crud.get_user_by_email(db, user.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db, user)

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, updates: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.update_user(db, user_id, updates.model_dump(exclude_unset=True))
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    deleted = crud.delete_user(db, user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="User not found")
    return

Run and verify

Start the server with uvicorn main:app --reload, open the interactive docs at http://localhost:8000/docs, and try the endpoints:

• POST /users/ with {"email": "alice@example.com", "hashed_password": "secret"} — returns the created user with an ID.
• GET /users/1 — returns the user you just created.
• PUT /users/1 with {"email": "alice.new@example.com"} — updates the email.
• DELETE /users/1 — removes the user; subsequent GET returns 404.

Expected output for the POST request:

{
  "id": 1,
  "email": "alice@example.com",
  "hashed_password": "secret"
}

Full CRUD integration

Notice how each endpoint uses the get_db dependency to get a session, calls the CRUD function, and then decides on the HTTP response. This separation makes your code testable and easy to extend.

Compare options / when to choose what

SQLAlchemy offers several ways to perform reads and writes. Choosing the right approach depends on your use case:

Notice: While SQLAlchemy 2.0 recommends select(), the session-based query() API is still supported and widely used. For this track, we use query() for simplicity and readability.

Troubleshooting & edge cases

Stale objects after commit

• Problem: After commit(), the object’s attributes (like auto-generated ID) may not be populated.
• Fix: Call db.refresh(obj) after committing to reload the object’s state from the database.

IntegrityError on duplicate entry

• Symptom: sqlalchemy.exc.IntegrityError when creating a user with an existing email.
• Fix: Catch the exception and roll back, or pre-check with get_user_by_email.

Session closed or expired attributes

• Symptom: DetachedInstanceError when accessing attributes after the session is closed.
• Fix: Access all needed attributes inside the session scope or use db.expunge() to detach, but beware of lazy loads.

Forgetting commit() or close()

• Consequence: Changes not persisted, or connection leak.
• Fix: Always commit after writes and close sessions in a finally block — the get_db dependency does this for you.

N+1 queries

• Symptom: Slow performance when fetching lists with relationships.
• Fix: Use joinedload or selectinload from sqlalchemy.orm to eager-load relationships.

What you learned & what's next

You now have a complete, reusable CRUD pattern with SQLAlchemy. You can:

• Create new records with add and commit
• Read single or multiple records with query and filters
• Update by modifying attributes and committing
• Delete with delete and commit
• Use FastAPI dependencies to manage sessions cleanly

This blueprint is the foundation for every persistent-backed REST API. Next, you’ll learn how to integrate SQLAlchemy models with Pydantic schemas for clean request/response validation and serialization — that’s the bridge between your database layer and your HTTP layer.