Databases with Flask-SQLAlchemy

Define models, query them through the session, handle transactions without leaving the session broken, and avoid the lazy-loading traps that make pages slow.

Models and the session

# app/models.py
from datetime import datetime, timezone
from werkzeug.security import generate_password_hash, check_password_hash
from .extensions import db

class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime(timezone=True),
                           default=lambda: datetime.now(timezone.utc))

    posts = db.relationship("Post", back_populates="author",
                            cascade="all, delete-orphan", lazy="select")

    def set_password(self, raw):
        self.password_hash = generate_password_hash(raw)

    def check_password(self, raw):
        return check_password_hash(self.password_hash, raw)

    def __repr__(self):
        return f"<User {self.email}>"

class Post(db.Model):
    __tablename__ = "posts"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    body = db.Column(db.Text, default="")
    published = db.Column(db.Boolean, default=False, nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)

    author = db.relationship("User", back_populates="posts")
    tags = db.relationship("Tag", secondary="post_tags", back_populates="posts")

Querying

from sqlalchemy import func, select
from app.extensions import db
from app.models import Post, User

db.session.execute(db.select(Post)).scalars().all()          # 2.x style
Post.query.filter_by(published=True).all()                   # legacy, still works

stmt = (db.select(Post)
          .where(Post.published.is_(True), Post.title.ilike("%flask%"))
          .order_by(Post.id.desc())
          .limit(20))

db.session.execute(stmt).scalars().all()
db.session.get(Post, 42)                                     # by primary key, None if absent
db.get_or_404(Post, 42)                                      # raises 404 inside a request
db.session.execute(db.select(Post).filter_by(id=42)).scalar_one_or_none()

# counts and aggregates run in the database
db.session.execute(db.select(func.count()).select_from(Post)).scalar_one()

# eager loading: one query for posts, one for their authors
stmt = (db.select(Post)
          .options(db.joinedload(Post.author), db.selectinload(Post.tags))
          .where(Post.published.is_(True)))
posts = db.session.execute(stmt).scalars().all()

# pagination: page 1, 20 per page, errors_outside=True to 404 on out-of-range
page = db.paginate(db.select(Post).order_by(Post.id), page=1, per_page=20, error_out=False)
page.items, page.total, page.pages, page.has_next

• The session is a unit-of-work and an identity map: query for the same row twice and you get the same Python object. It is created per request (Flask-SQLAlchemy scopes it to the app context), so never cache models across requests.
• .first() adds LIMIT 1 and returns None; .one() raises if there is not exactly one row; .scalar_one_or_none() is the explicit middle ground. Pick the one that encodes your intent.
• Accessing post.author inside a loop issues one query per row. The fix is joinedload for many-to-one and selectinload for one-to-many or many-to-many.
• db.paginate runs a COUNT plus a windowed SELECT. On a large table with a deep page offset, the count is the expensive half — cache it or use keyset pagination.

Transactions and constraints

from sqlalchemy.exc import IntegrityError
from flask import abort, flash, redirect, url_for

@app.post("/posts")
def create_post():
    form = request.form
    post = Post(title=form["title"].strip(), body=form.get("body", ""),
                author_id=current_user.id)
    db.session.add(post)
    try:
        db.session.commit()             # one commit per request, at the end
    except IntegrityError:
        db.session.rollback()           # mandatory before any further query
        flash("A post with that title already exists.", "error")
        return redirect(url_for("blog.new_post"))
    return redirect(url_for("blog.detail", post_id=post.id))

# add many rows, then one commit
db.session.add_all(Post(title=f"Seed {i}") for i in range(100))
db.session.commit()

# insert a lot of rows efficiently
db.session.execute(db.insert(Post), [{"title": "a"}, {"title": "b"}])
db.session.commit()

# you need the primary key before the commit? flush, do not commit
db.session.add(post)
db.session.flush()                      # sends INSERT, keeps the transaction open
post.id                                 # now populated

• A failed commit leaves the session unusable. Any statement after an IntegrityError without a rollback() raises PendingRollbackError, which hides the original problem in the traceback.
• Keep transactions short. Holding one open across a network call or a template render holds locks, and on SQLite it blocks every other writer.
• Translate a unique-constraint violation into the right HTTP answer: a duplicate resource is 409 Conflict, a bad value is 422 or 400. Letting the exception reach the 500 handler tells the client nothing.

FAQ

Related

Schema migrations with Flask-Migrate Authentication with Flask-Login and password hashing

Last refreshed 2026-09-18.