Flask + SQLAlchemy: Real Persistence — Web Frameworks

In-memory dicts vanish when the server restarts. Real apps need a database — somewhere data lives between requests, surfaces in queries, and survives a deploy. Flask-SQLAlchemy is the standard integration: the Flask glue around SQLAlchemy 2.0, the modern Python ORM.

This lesson builds on databases (the stdlib sqlite3 + raw-SQLAlchemy introduction) and adds the Flask-specific pieces: app-bound configuration, the modern declarative model syntax, migrations with Flask-Migrate, pagination, relationship loading, and the N+1 problem. By the end, you'll replace the dict from flask-basics with a real database.

Run locally with pip install flask flask-sqlalchemy flask-migrate. SQLite needs nothing else; for Postgres add pip install psycopg[binary] and set DATABASE_URL to a postgresql+psycopg://... URI. Expected output is shown in comments.

1. Setup

bash

pip install flask flask-sqlalchemy flask-migrate

Configuration goes in the Flask app config:

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

db = SQLAlchemy(model_class=Base)

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False    # silence a deprecation warning
db.init_app(app)

Three settings to know:

SQLALCHEMY_DATABASE_URI — the connection string. SQLite: sqlite:///app.db (relative) or sqlite:////absolute/path.db (four slashes for absolute). Postgres: postgresql+psycopg://user:pass@host:5432/dbname. Read it from an env var in production — see devops-secrets.
SQLALCHEMY_TRACK_MODIFICATIONS = False — turn off a legacy feature most apps don't use. Avoids a startup warning.
SQLALCHEMY_ECHO = True — useful during development; echoes every SQL statement to stdout. Off in production.

In real projects you create db at module level and call db.init_app(app) inside a factory function. The pattern is covered in flask-blueprints; for this lesson the simpler global-app style is fine.

2. Defining Models — Modern 2.0 Style

SQLAlchemy 2.0 introduced typed declarative models using Mapped[T] and mapped_column(). The types are real PEP 484 annotations — your IDE, mypy, and Pyright understand them.

python

from datetime import datetime
from sqlalchemy import String, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

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

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(80))
    active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # Relationship — list of posts authored by this user
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

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


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

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column()                        # Text in the DB
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    author: Mapped["User"] = relationship(back_populates="posts")

+ setup added so this can run · defines db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

db = _AutoMock('db')

Key points:

Mapped[int] — the type annotation drives column inference. int → Integer, str → String/Text, bool → Boolean, datetime → DateTime. Override with the first positional argument: String(120) for a length cap.
mapped_column(primary_key=True) — column-level options. Common ones: unique, nullable, default (Python-side), server_default (database-side), index.
relationship(back_populates=...) — defines the in-Python attribute that follows a foreign key. back_populates wires the two sides together (so user.posts and post.author stay in sync in memory).
cascade="all, delete-orphan" — when a user is deleted, their posts go with them. Without it, deleting a user with posts raises an IntegrityError.

Older SQLAlchemy code uses Column(Integer, primary_key=True) instead of mapped_column. Both work; the new style is strictly better — type-aware and less repetitive. New code should use it.

3. Creating Tables

For a quick start, ask SQLAlchemy to create everything declared:

python

with app.app_context():
    db.create_all()

+ setup added so this can run · defines app, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
db = _AutoMock('db')

Why the app_context()? Flask-SQLAlchemy looks up the bound app to find the engine. Inside a request handler it's automatic; in a one-off script you have to push the context yourself.

db.create_all() is for development and demos. It creates tables that don't exist; it doesn't alter existing ones. The moment you add a column to a model, create_all() silently does nothing. For real apps you use migrations (Section 9).

4. CRUD in a Route

python

from sqlalchemy import select

@app.post("/users")
def create_user():
    data = request.get_json()
    user = User(email=data["email"], name=data["name"])
    db.session.add(user)
    db.session.commit()                              # without this, the INSERT vanishes
    return {"id": user.id, "email": user.email}, 201


@app.get("/users/<int:user_id>")
def get_user(user_id):
    user = db.session.get(User, user_id)              # PK lookup — by primary key
    if user is None:
        abort(404)
    return {"id": user.id, "email": user.email, "name": user.name}


@app.patch("/users/<int:user_id>")
def update_user(user_id):
    user = db.session.get(User, user_id) or abort(404)
    data = request.get_json()
    if "name" in data:
        user.name = data["name"]
    db.session.commit()                              # no add() needed — already tracked
    return {"id": user.id, "name": user.name}


@app.delete("/users/<int:user_id>")
def delete_user(user_id):
    user = db.session.get(User, user_id) or abort(404)
    db.session.delete(user)
    db.session.commit()
    return "", 204

+ setup added so this can run · defines User, app, request, abort, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def User(*_a, **_kw):
    print('-> User() called')
    return _AutoMock('User()')
app = _AutoMock('app')
request = _AutoMock('request')
def abort(*_a, **_kw):
    print('-> abort() called')
    return _AutoMock('abort()')
db = _AutoMock('db')

The db.session is a per-request transactional context. Three operations matter:

db.session.add(obj) — stage a new row for INSERT.
db.session.delete(obj) — stage a row for DELETE.
db.session.commit() — flush all staged changes and commit the transaction. Without it, your changes vanish at the end of the request.

For updates, you don't call add — modifying an attribute on a loaded object marks it dirty, and commit() emits the UPDATE.

If anything raises before commit(), Flask-SQLAlchemy automatically rolls back at the end of the request. You don't have to wrap routes in try/except — but you do if you want to handle specific errors like IntegrityError.

5. Querying — The Modern Style

SQLAlchemy 2.0 unified query syntax around select(). The old User.query.filter(...) style still works for Flask-SQLAlchemy (it's a back-compat shim), but select(...) is the future:

python

from sqlalchemy import select

# All active users
stmt = select(User).where(User.active == True).order_by(User.created_at.desc())
users = db.session.execute(stmt).scalars().all()

# One user by email — returns None if missing
stmt = select(User).where(User.email == "margaret@example.com")
user = db.session.execute(stmt).scalar_one_or_none()

# Expecting exactly one — raises if zero or more than one
user = db.session.execute(stmt).scalar_one()

# Just the first result
user = db.session.execute(stmt).scalars().first()

# Count
from sqlalchemy import func
total = db.session.execute(select(func.count()).select_from(User)).scalar_one()

+ setup added so this can run · defines User, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

User = _AutoMock('User')
db = _AutoMock('db')

Decoder ring for the chain:

select(User) — build a SELECT statement.
.where(...), .order_by(...), .limit(...), .offset(...) — modifiers.
db.session.execute(stmt) — run it, returns a Result object.
.scalars() — when you selected whole entities (select(User)), unwrap from the row tuple. Without it you get Row objects (tuples).
.all() / .first() / .one() / .one_or_none() / .scalar_one() — terminal methods that materialise rows.

The legacy style, still common in older codebases:

python

# Legacy — works, but not the recommended 2.0 style
users = User.query.filter_by(active=True).order_by(User.created_at.desc()).all()
user = User.query.filter_by(email="margaret@example.com").first()
user = User.query.get(42)

+ setup added so this can run · defines User

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

User = _AutoMock('User')

Both produce the same SQL. New code should write select(...).

6. Pagination

Flask-SQLAlchemy 3.x ships a paginate() helper that wraps a SELECT with LIMIT/OFFSET and useful metadata:

python

@app.get("/users")
def list_users():
    page = request.args.get("page", 1, type=int)
    per_page = min(request.args.get("per_page", 20, type=int), 100)   # cap server-side

    stmt = select(User).order_by(User.created_at.desc())
    pagination = db.paginate(stmt, page=page, per_page=per_page, error_out=False)

    return {
        "items": [{"id": u.id, "email": u.email} for u in pagination.items],
        "page": pagination.page,
        "pages": pagination.pages,
        "total": pagination.total,
        "has_next": pagination.has_next,
        "has_prev": pagination.has_prev,
    }

+ setup added so this can run · defines app, db, request, select, User

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
db = _AutoMock('db')
request = _AutoMock('request')
def select(*_a, **_kw):
    print('-> select() called')
    return _AutoMock('select()')
User = _AutoMock('User')

Three things to lock down:

Cap per_page server-side. A malicious client sending ?per_page=1000000 will exhaust memory.
error_out=False — by default, requesting page 999 of a 10-page list raises 404. Set False and return an empty list for nicer API ergonomics.
Stable ordering. Without an ORDER BY, the same query can return different rows on different pages. Always order by something unique (usually id or created_at, id).

For very large tables, OFFSET gets slow — the DB still has to scan every skipped row. Switch to keyset pagination (a.k.a. seek pagination): WHERE created_at < :last_seen ORDER BY created_at DESC LIMIT 20. Worth the upgrade past a few hundred thousand rows.

7. Relationships

The User.posts / Post.author pair from Section 2 is the canonical one-to-many. Use it like a regular Python attribute:

python

user = db.session.get(User, 1)
for post in user.posts:                             # one extra SELECT to load posts
    print(post.title)

# The other way — from a post to its author
post = db.session.get(Post, 42)
print(post.author.email)                            # one extra SELECT for the author

+ setup added so this can run · defines User, Post, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

User = _AutoMock('User')
Post = _AutoMock('Post')
db = _AutoMock('db')

Many-to-many uses an association table — a third table holding only the two foreign keys:

python

from sqlalchemy import Table, Column

post_tags = Table(
    "post_tags",
    db.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class Tag(db.Model):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    posts: Mapped[list["Post"]] = relationship(secondary=post_tags, back_populates="tags")

class Post(db.Model):
    # ... previous fields ...
    tags: Mapped[list["Tag"]] = relationship(secondary=post_tags, back_populates="posts")

+ setup added so this can run · defines db, Mapped, mapped_column, relationship, ForeignKey, String

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

db = _AutoMock('db')
Mapped = _AutoMock('Mapped')
def mapped_column(*_a, **_kw):
    print('-> mapped_column() called')
    return _AutoMock('mapped_column()')
def relationship(*_a, **_kw):
    print('-> relationship() called')
    return _AutoMock('relationship()')
def ForeignKey(*_a, **_kw):
    print('-> ForeignKey() called')
    return _AutoMock('ForeignKey()')
def String(*_a, **_kw):
    print('-> String() called')
    return _AutoMock('String()')

Now post.tags.append(tag) updates both sides; db.session.commit() writes the row to post_tags. No manual INSERT into the join table.

8. The N+1 Problem — And How to Avoid It

The most common ORM performance bug:

python

# WRONG — 1 query for the posts, then N queries (one per post) for each author
stmt = select(Post).limit(50)
posts = db.session.execute(stmt).scalars().all()
for p in posts:
    print(f"{p.title} — by {p.author.name}")        # triggers SELECT * FROM users WHERE id = ?

+ setup added so this can run · defines select, Post, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def select(*_a, **_kw):
    print('-> select() called')
    return _AutoMock('select()')
Post = _AutoMock('Post')
db = _AutoMock('db')

For 50 posts, that's 51 round-trips. On a fast local DB you barely notice. On a remote Postgres with 5ms latency, that's a quarter-second per page just on database time.

The fix — eager loading:

python

from sqlalchemy.orm import joinedload, selectinload

# Option A — joinedload: single query with a JOIN
stmt = select(Post).options(joinedload(Post.author)).limit(50)

# Option B — selectinload: one extra query that batches the IDs (better for collections)
stmt = select(User).options(selectinload(User.posts)).limit(50)

+ setup added so this can run · defines select, Post, User

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def select(*_a, **_kw):
    print('-> select() called')
    return _AutoMock('select()')
Post = _AutoMock('Post')
User = _AutoMock('User')

Rules of thumb:

joinedload — many-to-one (Post → Author). One JOIN, one query.
selectinload — one-to-many or many-to-many (User → Posts). Two queries total: SELECT ... FROM users then SELECT ... FROM posts WHERE author_id IN (...). Avoids the row explosion that JOINs create when each parent has many children.

Watch for N+1 specifically when:

You iterate a list of objects and touch a relationship inside the loop.
A template renders a list and accesses item.<relationship>.<field>.
You serialise objects to JSON and a to_dict method follows relationships.

Turn on SQLALCHEMY_ECHO=True during development and watch the query log when you load a page. Anything that emits hundreds of identical SELECTs is N+1.

9. Migrations with Flask-Migrate

db.create_all() is fine until the first time you add a column. Then you need migrations — versioned scripts that ALTER your existing schema. Flask-Migrate wraps Alembic, the standard tool:

python

from flask_migrate import Migrate

migrate = Migrate(app, db)

+ setup added so this can run · defines app, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')
db = _AutoMock('db')

Workflow:

bash

flask db init                                       # one-time — creates the migrations/ folder
flask db migrate -m "create users and posts"        # generates a migration script from model diffs
flask db upgrade                                    # applies pending migrations to the DB

# Later, after adding a column:
flask db migrate -m "add User.bio"
flask db upgrade

# Roll back the last one if you broke something:
flask db downgrade

The generated script in migrations/versions/... is Python you should read before running. Alembic auto-detects most changes (new tables, new columns, dropped columns, index changes) but it can't see semantic changes — renames look like a drop + add by default, which would destroy data. Edit the script when needed.

Migrations belong in version control. Every developer and every environment runs the same scripts in the same order — that's the whole guarantee.

10. Transactions

db.session is transactional by default. Within a single request, all your adds and deletes and attribute changes pile up; commit() flushes them atomically. If anything raises before commit(), Flask-SQLAlchemy rolls back automatically.

For multi-step operations where you want explicit control:

python

try:
    user = User(email=email, name=name)
    db.session.add(user)
    db.session.flush()                              # gets user.id without committing

    profile = Profile(user_id=user.id, bio="")
    db.session.add(profile)
    db.session.commit()                             # both rows committed together, or neither
except IntegrityError:
    db.session.rollback()
    raise ConflictError("email already exists")

+ setup added so this can run · defines IntegrityError, User, Profile, email, name, ConflictError, db

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

IntegrityError = _AutoMock('IntegrityError')
def User(*_a, **_kw):
    print('-> User() called')
    return _AutoMock('User()')
def Profile(*_a, **_kw):
    print('-> Profile() called')
    return _AutoMock('Profile()')
email = _AutoMock('email')
name = _AutoMock('name')
def ConflictError(*_a, **_kw):
    print('-> ConflictError() called')
    return _AutoMock('ConflictError()')
db = _AutoMock('db')

flush() sends SQL to the DB but stays inside the transaction. Useful when you need a newly-generated primary key (like user.id above) before committing.

11. Connection Pooling for Production

Each db.session.execute(...) borrows a connection from a pool, runs SQL, returns the connection. The pool is sized by config:

python

app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_size": 10,                                # connections kept open
    "max_overflow": 5,                              # extra connections under burst
    "pool_pre_ping": True,                          # check connections before use — avoids stale-conn errors
    "pool_recycle": 300,                            # recycle every 5 min — handle idle timeouts
}

+ setup added so this can run · defines app

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

app = _AutoMock('app')

pool_pre_ping=True is the line that saves you at 3 AM when a network blip drops idle connections — without it, the first request after a blip fails with a confusing error. It costs one extra round-trip per connection acquisition; in production that's always worth it.

For Postgres, set pool_size so that pool_size * num_workers <= max_connections on the DB. Otherwise a deploy under load will exhaust the Postgres connection limit. See db-postgres for the production-database deep dive.

12. Common Mistakes

1. Forgetting db.session.commit()

Symptom: your route returns 200, but the data isn't there on reload. The transaction rolled back at request end. Add commit(). (Many tutorials helpfully wire up @app.teardown_request to auto-commit — don't. Explicit commits make the transaction boundary visible.)

2. Querying inside templates

html

{# DANGER #}
{% for post in posts %}
    <p>{{ post.title }} by {{ post.author.name }}</p>      {# N+1 — lazy-loads each author #}
{% endfor %}

Templates are an easy place to accidentally trigger queries — post.author looks like attribute access, but it's a DB hit. Eager-load relationships in the view (selectinload, joinedload) so the template just reads in-memory data.

3. Shipping with SQLite to production

SQLite is brilliant for development, tests, single-user apps. It serialises writes globally. The moment you have a second process (e.g. running gunicorn --workers 4), writes start blocking each other and you'll see random database is locked errors. Use Postgres for anything multi-process. See db-postgres.

4. Not handling IntegrityError on unique constraints

python

@app.post("/users")
def create_user():
    user = User(email=request.json["email"])
    db.session.add(user)
    db.session.commit()                             # raises IntegrityError if email exists

+ setup added so this can run · defines User, app, db, request

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def User(*_a, **_kw):
    print('-> User() called')
    return _AutoMock('User()')
app = _AutoMock('app')
db = _AutoMock('db')
request = _AutoMock('request')

Catch it and return a clean 409:

python

from sqlalchemy.exc import IntegrityError

try:
    db.session.add(user)
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    return {"error": "email already in use"}, 409

Or check existence first (with a race-condition caveat) — usually you do both: check for the friendly UX, catch IntegrityError for the race.

5. Auto-generated migration applied without reading it

Alembic guesses. Sometimes it guesses wrong — column renames become drop+add (data loss), type changes can lose precision, index changes can be expensive on big tables. Always read the generated migration before running it. Always.

6. Mixing User.query legacy syntax with new select() style randomly

Pick one for new code. The User.query.filter(...) shim still works in Flask-SQLAlchemy 3.x for backward compatibility, but the 2.0 select() style is the future and the docs assume it. Consistency in a codebase > strict modernity.

🎯 Your Turn — Persist the Items App

Take the items app from flask-basics and replace the in-memory dict with a real database using Flask-SQLAlchemy and Flask-Migrate.

Requirements:

An Item model with id, name (1-100 chars, required), created_at.
GET / lists items ordered by newest first.
GET /items/<int:item_id> shows one item, 404 if missing.
GET /items/new shows a form; POST /items/new creates an item and redirects.
Use Flask-Migrate — initialise migrations, generate the first migration, run it.
Use select(...) style queries, not the legacy Item.query.

Skeleton:

python

# app.py
from datetime import datetime
from flask import Flask, request, redirect, url_for, render_template, abort
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy import select, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

db = SQLAlchemy(model_class=Base)
migrate = Migrate()

# TODO 1: define class Item(db.Model) with id, name, created_at

def create_app():
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///items.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.init_app(app)
    migrate.init_app(app, db)

    # TODO 2: register the three routes (home, show, new GET, new POST)

    return app

app = create_app()

After writing the code, run:

bash

flask db init
flask db migrate -m "create items"
flask db upgrade
flask run

Hint 1 — Defining the Item model

class Item(db.Model):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

Note String(100) caps the column width at the database level — your form validator should match.

Hint 2 — Querying with select()

For the home page list:

stmt = select(Item).order_by(Item.created_at.desc())
items = db.session.execute(stmt).scalars().all()

For a single item by ID, prefer db.session.get(Item, item_id) — it uses the identity-map cache and returns None if missing, perfect for an abort(404).

Show full solution

python

# app.py
from datetime import datetime
from flask import Flask, request, redirect, url_for, render_template, abort, flash
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy import select, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


db = SQLAlchemy(model_class=Base)
migrate = Migrate()


class Item(db.Model):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    def __repr__(self) -> str:
        return f"<Item {self.id} {self.name!r}>"


def create_app():
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "dev-only"
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///items.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.init_app(app)
    migrate.init_app(app, db)

    @app.get("/")
    def home():
        stmt = select(Item).order_by(Item.created_at.desc())
        items = db.session.execute(stmt).scalars().all()
        return render_template("index.html", items=items)

    @app.get("/items/<int:item_id>")
    def show_item(item_id):
        item = db.session.get(Item, item_id)
        if item is None:
            abort(404)
        return render_template("detail.html", item=item)

    @app.get("/items/new")
    def new_item_form():
        return render_template("new.html")

    @app.post("/items/new")
    def create_item():
        name = (request.form.get("name") or "").strip()
        if not name or len(name) > 100:
            flash("Name is required and must be 1-100 characters.", "error")
            return redirect(url_for("new_item_form"))
        item = Item(name=name)
        db.session.add(item)
        db.session.commit()
        flash(f"Created '{item.name}'.", "success")
        return redirect(url_for("home"))

    @app.errorhandler(404)
    def not_found(_e):
        return render_template("404.html"), 404

    return app


app = create_app()

if __name__ == "__main__":
    app.run(debug=True)

Migration commands you run once after writing the model:

bash

export FLASK_APP=app.py          # PowerShell: $env:FLASK_APP="app.py"
flask db init                    # one-time
flask db migrate -m "create items"
flask db upgrade
flask run

What this gets right:

2.0-style models with Mapped[...] and mapped_column(...) — type-checked, IDE-friendly.
server_default=func.now() — the DB sets created_at, not Python. Survives clock drift between app servers and the DB.
select(Item).order_by(...) + scalars().all() — the modern query idiom.
db.session.get(Item, id) for primary-key lookup — uses the identity cache, no SQL if already loaded.
flash() + POST-redirect-GET — refresh after submit doesn't re-insert.
Migrations from day one — flask db init / migrate / upgrade. The next column you add is one migration away.

What's missing for production:

Form validation with Flask-WTF — covered in flask-forms. The hand-rolled validation above is fine for one field; it doesn't scale.
Postgres in production — SQLite serialises writes globally. See db-postgres.
Tests — pytest with app.test_client() and a fresh in-memory SQLite per test. Covered in flask-blueprints.
The Item route handlers belong in a blueprint once you add more resources. Next lesson.

What You Learned

Flask-SQLAlchemy wraps SQLAlchemy 2.0 with Flask integration. Configure with SQLALCHEMY_DATABASE_URI, initialise with db.init_app(app).
Modern models use Mapped[T] + mapped_column(...) — typed, IDE-friendly, the 2.0 default.
relationship(back_populates=...) wires the two sides of a foreign key. cascade="all, delete-orphan" ties child lifetimes to parents.
CRUD: db.session.add(obj), db.session.delete(obj), attribute mutation for updates, db.session.commit() to flush. Auto-rollback on request error.
Queries: select(Model).where(...).order_by(...) → db.session.execute(stmt).scalars() → .all() / .first() / .scalar_one_or_none().
db.paginate(stmt, page=..., per_page=...) — pagination with metadata; cap per_page server-side; always order by something unique.
The N+1 problem kills performance. Eager-load with joinedload (many-to-one) or selectinload (one-to-many / many-to-many).
Migrations with Flask-Migrate: flask db init, migrate -m "...", upgrade. Always read the generated script before running.
Connection pooling for production: pool_pre_ping=True, sensible pool_size. Use Postgres, not SQLite, for multi-process deployments.
db.create_all() is for demos. Real apps use migrations from the first commit.

Next: flask-blueprints — splitting one big app.py into a create_app() factory + per-feature blueprints, plus testing and production deployment.