Flask + SQLAlchemy: Real Persistence — Web Frameworks

In-memory dicts vanish when the server restarts. Real apps need a database: a place where data stays between requests, can be queried, and survives a redeploy. Flask-SQLAlchemy is the standard way to connect them: it adds Flask support to 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: turns off an old feature that most apps don't use. This also avoids a startup warning.
SQLALCHEMY_ECHO = True: prints every SQL statement to stdout. This is useful during development. Turn it 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, so 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: the list of posts written 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]: SQLAlchemy chooses the column type from the type annotation. 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 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 are deleted too. 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 better: it knows the types and needs less repetition. 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, Flask does this for you. 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, but it doesn't alter existing ones. If you add a column to an existing model, create_all() does not add it, and it shows no error. 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 commit(), the INSERT is lost
    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)              # look up the row by its 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, because the session already tracks user
    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 are lost at the end of the request.

For updates, you don't call add. When you change an attribute on a loaded object, the session marks the object as changed ("dirty"), and commit() sends 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. You only need it 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 in Flask-SQLAlchemy (it is kept for backward compatibility), but select(...) is the recommended style:

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 there is no match)
stmt = select(User).where(User.email == "margaret@example.com")
user = db.session.execute(stmt).scalar_one_or_none()

# Exactly one row expected. It raises an error for zero rows 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')

What each part of the chain does:

select(User): build a SELECT statement.
.where(...), .order_by(...), .limit(...), .offset(...): modifiers.
db.session.execute(stmt): runs the statement and returns a Result object.
.scalars(): when you selected whole entities (select(User)), this takes each object out of its row tuple. Without it you get Row objects (tuples).
.all() / .first() / .one() / .one_or_none() / .scalar_one(): final methods that fetch the rows.

The legacy style, still common in older codebases:

python

# Legacy style: it works, but it is 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)   # the server caps it at 100

    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 get right:

Cap per_page server-side. A malicious client can send ?per_page=1000000 and use up the server's memory.
error_out=False: by default, requesting page 999 of a 10-page list raises 404. Set it to False to return an empty list instead, which is easier for API clients.
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, because the DB still has to scan every skipped row. Switch to keyset pagination (also called seek pagination): WHERE created_at < :last_seen ORDER BY created_at DESC LIMIT 20. It is worth the change once a table has more than a few hundred thousand rows.

7. Relationships

The User.posts / Post.author pair from Section 2 is the standard one-to-many example. 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 direction: 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 that holds 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, and db.session.commit() writes the row to post_tags. You don't write the INSERT into the join table yourself.

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 more queries (one per post) for the authors
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 about a quarter of a second per page, just waiting for the database.

The fix is eager loading:

python

from sqlalchemy.orm import joinedload, selectinload

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

# Option B: selectinload makes one extra query for all 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 (...). This avoids the very large result that a JOIN creates when each parent has many children (each parent row is repeated once per child).

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                                       # run once, to create the migrations/ folder
flask db migrate -m "create users and posts"        # writes a migration script from your model changes
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 code, and you should read it before running it. Alembic detects most changes (new tables, new columns, dropped columns, index changes), but it can't tell what you meant. By default, a rename looks like a drop plus an add, which would delete the data in that column. Edit the script when needed.

Migrations belong in version control. Then every developer and every environment runs the same scripts in the same order, so every database ends up with the same schema.

10. Transactions

db.session is transactional by default. Within a single request, all your adds, deletes and attribute changes are collected, and commit() writes them all in one transaction. 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. This is 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 allowed during a burst of traffic
    "pool_pre_ping": True,                          # test each connection before use, to avoid errors from stale connections
    "pool_recycle": 300,                            # replace connections after 5 minutes, before the DB closes idle ones
}

+ 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 matters when a short network problem closes idle connections. Without it, the first request after the problem fails with a confusing error. It costs one extra round-trip each time a connection is taken from the pool, and in production that cost is worth paying.

For Postgres, set pool_size so that pool_size * num_workers <= max_connections on the DB. Otherwise, under load, your app can use up all the connections Postgres allows. See db-postgres for more on running a production database.

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(). (Some tutorials use @app.teardown_request to commit automatically. Don't do this. An explicit commit() shows clearly where the transaction ends.)

2. Querying inside templates

html

{# DANGER #}
{% for post in posts %}
    <p>{{ post.title }} by {{ post.author.name }}</p>      {# N+1 problem: each author is loaded with its own query #}
{% endfor %}

It is easy to trigger queries by accident in a template. post.author looks like normal attribute access, but it runs a database query. Eager-load relationships in the view (selectinload, joinedload) so the template just reads in-memory data.

3. Shipping with SQLite to production

SQLite is very good for development, tests and single-user apps. But it allows only one write at a time for the whole database. As soon as 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

You can also check whether the row exists first, but two requests can still pass that check at the same time (a race condition). Usually you do both: check first to show a friendly message, and catch IntegrityError for the race.

5. Auto-generated migration applied without reading it

Alembic guesses, and 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.

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 recommended one and the docs use it. In an existing codebase, being consistent matters more than using the newest style.

🎯 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, and 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) limits the column length at the database level. Your form validator should use the same limit.

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 the item is missing, which works well with 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                    # only needed once
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 and IDE-friendly.
server_default=func.now(): the DB sets created_at, not Python. So the time stays correct even if the app servers' clocks differ from the DB's clock.
select(Item).order_by(...) + scalars().all(): the modern way to write a query.
db.session.get(Item, id) for primary-key lookup: it uses the identity cache, so no SQL runs if the item is already loaded.
flash() + POST-redirect-GET: refreshing the page after submitting doesn't insert the item again.
Migrations from the start: flask db init / migrate / upgrade. To add the next column, you only need one new migration.

What's missing for production:

Form validation with Flask-WTF: covered in flask-forms. The hand-written validation above is fine for one field, but it gets hard to manage with many fields.
Postgres in production: SQLite allows only one write at a time. 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, and 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=...) gives pagination with metadata. Cap per_page server-side, and always order by something unique.
The N+1 problem makes pages slow. 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.