🔁 CRUD Operations with SQLAlchemy

What CRUD Means

CRUD stands for Create, Read, Update, Delete — the four fundamental operations on stored data. Almost every feature you build is some combination of these four: signing up is a Create, a profile page is a Read, editing settings is an Update, closing an account is a Delete.

💡 A useful analogy: Picture a library. Create is shelving a new book; Read is looking one up in the catalog; Update is correcting its record; Delete is removing it for good. Flask-SQLAlchemy is the librarian who does all four for you when you ask in Python.

The Session

Every CRUD operation goes through db.session. The session is a staging area: you add, modify, and mark objects for deletion, and nothing actually hits the database until you call commit(). This is the unit of work pattern — a batch of changes applied as one all-or-nothing transaction.

Create

Creating a record is three steps: build the object, add it, commit. After the commit, database-generated values like the auto-increment id are populated back onto your object.

user = User(username="john_doe", email="john@example.com")
db.session.add(user)
db.session.commit()

print(user.id)   # now has a real value from the database

Creating related records

Relationships let you link objects without manually juggling foreign keys. Append a child to a parent's collection and SQLAlchemy sets the foreign key for you on commit.

user = User(username="jane_doe", email="jane@example.com")
post = Post(title="My First Post", content="Hello, world!")

# The relationship sets post.user_id automatically
user.posts.append(post)

db.session.add(user)   # adding the parent cascades to the new child
db.session.commit()

Bulk inserts

users = [
    User(username=f"user{i}", email=f"user{i}@example.com")
    for i in range(1, 4)
]
db.session.add_all(users)
db.session.commit()

Read & Query

SQLAlchemy 2.0 unifies querying around db.select() plus db.session.execute(). The older Model.query.all() style still works in Flask-SQLAlchemy, but the select() style is the modern, recommended one and reads the same whether you're in Flask or plain SQLAlchemy.

Fetching rows

# All users
users = db.session.execute(db.select(User)).scalars().all()

# The first match
user = db.session.execute(db.select(User)).scalars().first()

# By primary key
user = db.session.get(User, 1)          # returns None if missing

# By primary key or 404 (handy in a route)
user = db.get_or_404(User, 1)

Filtering, ordering, limiting

from sqlalchemy import or_

# WHERE is_admin = true
stmt = db.select(User).where(User.is_admin.is_(True))
admins = db.session.execute(stmt).scalars().all()

# Multiple conditions are ANDed together
stmt = db.select(User).where(User.is_active.is_(True), User.age >= 18)

# OR
stmt = db.select(User).where(
    or_(User.username == "admin", User.email.endswith("@admin.com"))
)

# Case-insensitive search + newest first + only 10 rows
stmt = (
    db.select(User)
    .where(User.username.ilike("%john%"))
    .order_by(User.created_at.desc())
    .limit(10)
)
results = db.session.execute(stmt).scalars().all()

Pagination

Flask-SQLAlchemy's db.paginate() turns a select statement into a page of results plus navigation metadata — exactly what a "page 3 of 27" listing needs.

page = request.args.get("page", 1, type=int)
pagination = db.paginate(
    db.select(User).order_by(User.username),
    page=page,
    per_page=10,
)

pagination.items      # the rows on this page
pagination.total      # total row count
pagination.pages      # total number of pages
pagination.has_next   # is there a next page?
pagination.next_num   # its page number

Aggregation

from sqlalchemy import func

# A simple count
total = db.session.scalar(db.select(func.count()).select_from(User))

# Count posts per user, most prolific first
stmt = (
    db.select(User.username, func.count(Post.id).label("post_count"))
    .join(Post)
    .group_by(User.id)
    .order_by(func.count(Post.id).desc())
)
rows = db.session.execute(stmt).all()   # list of (username, post_count) tuples

Update

To update one record, load it, change its attributes, and commit. SQLAlchemy tracks which attributes changed and issues the minimal UPDATE.

user = db.session.get(User, 1)
user.username = "new_username"
user.email = "new_email@example.com"
db.session.commit()

Bulk updates

To change many rows without loading them all into memory, use an update() statement.

from sqlalchemy import update

stmt = (
    update(User)
    .where(User.is_active.is_(False))
    .values(last_login=None)
)
db.session.execute(stmt)
db.session.commit()

stmt = (
    update(Post)
    .where(Post.id == post_id)
    .values(view_count=Post.view_count + 1)
)
db.session.execute(stmt)
db.session.commit()

Delete

Deleting one record mirrors updating: load it, mark it for deletion, commit.

user = db.session.get(User, 1)
db.session.delete(user)
db.session.commit()

Cascade deletes

By default, deleting a parent does not delete its children — you could be left with orphaned posts pointing at a user who no longer exists. Configure the cascade on the relationship to delete children automatically.

class User(db.Model):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    # Deleting a user now deletes all of their posts too
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",
    )

class SoftDeleteMixin:
    deleted_at: Mapped[datetime | None] = mapped_column(default=None)

    def soft_delete(self) -> None:
        self.deleted_at = datetime.now(timezone.utc)

    @property
    def is_deleted(self) -> bool:
        return self.deleted_at is not None

Transactions & Performance

Handle errors with try / rollback

If any step in a multi-step write fails, roll the whole thing back so you never persist a half-finished change.

from sqlalchemy.exc import SQLAlchemyError

try:
    user = User(username="taylor", email="taylor@example.com")
    db.session.add(user)

    post = Post(title="Hello", content="First post!", author=user)
    db.session.add(post)

    db.session.commit()            # both rows, or neither
except SQLAlchemyError:
    db.session.rollback()          # undo everything staged
    raise                          # let the caller / error handler decide

Avoid the N+1 query problem

Relationships load lazily by default: touch user.posts and SQLAlchemy fires a query. Loop over 100 users touching their posts and you've fired 101 queries — one for the users, one per user. That's the N+1 problem, and it's the single most common cause of a slow ORM-backed page.

from sqlalchemy.orm import selectinload

# BAD: 1 query for users, then 1 more per user = N+1 queries
users = db.session.execute(db.select(User).limit(100)).scalars().all()
for u in users:
    print(len(u.posts))   # each access triggers another query

# GOOD: 2 queries total, no matter how many users
stmt = db.select(User).options(selectinload(User.posts)).limit(100)
users = db.session.execute(stmt).scalars().all()
for u in users:
    print(len(u.posts))   # already loaded — no extra queries

Hands-on Exercise

from flask import Blueprint, render_template, request, redirect, url_for, flash
from sqlalchemy.exc import SQLAlchemyError
from app.extensions import db
from app.models import Post

bp = Blueprint("posts", __name__)


@bp.route("/")
def home():
    # READ (list)
    stmt = db.select(Post).order_by(Post.created_at.desc())
    posts = db.session.execute(stmt).scalars().all()
    return render_template("home.html", posts=posts)


@bp.route("/post/<int:post_id>")
def detail(post_id):
    # READ (one)
    post = db.get_or_404(Post, post_id)
    return render_template("detail.html", post=post)


@bp.route("/post/new", methods=["GET", "POST"])
def create():
    if request.method == "POST":
        title = request.form.get("title", "").strip()
        content = request.form.get("content", "").strip()
        if not title or not content:
            flash("Title and content are required.", "danger")
            return render_template("form.html")

        post = Post(title=title, content=content)
        db.session.add(post)
        try:
            db.session.commit()               # CREATE
            flash("Post created!", "success")
            return redirect(url_for("posts.detail", post_id=post.id))
        except SQLAlchemyError:
            db.session.rollback()
            flash("Could not save the post.", "danger")
    return render_template("form.html")


@bp.route("/post/<int:post_id>/delete", methods=["POST"])
def delete(post_id):
    post = db.get_or_404(Post, post_id)
    try:
        db.session.delete(post)               # DELETE
        db.session.commit()
        flash("Post deleted.", "success")
    except SQLAlchemyError:
        db.session.rollback()
        flash("Could not delete the post.", "danger")
    return redirect(url_for("posts.home"))

Summary & Quiz

📚 Further Reading

• Flask-SQLAlchemy — Querying & Pagination
• SQLAlchemy 2.0 — ORM Querying Guide
• SQLAlchemy 2.0 — Session Basics

🚀 What's Next?

You can now store and manipulate data through Flask-SQLAlchemy. Next we'll formalize the structure you've been hinting at with the Application Factory Pattern — the professional way to organize a growing Flask project.