Intermediate 16 min

Databases with SQLAlchemy

Models, queries and migrations with Flask-SQLAlchemy.

What you will learn

Define models
Query and commit
Understand migrations

Most apps need a database. The usual choice with Flask is Flask-SQLAlchemy, which wraps SQLAlchemy, Python's leading ORM, so you work with classes instead of raw SQL.

pip install flask flask-sqlalchemy flask-migrate

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timezone

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False)
    name = db.Column(db.String(80), nullable=False)
    posts = db.relationship("Post", backref="author", lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    body = db.Column(db.Text)
    created = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)

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

with app.app_context():
    ada = User(email="ada@example.com", name="Ada")
    db.session.add(ada)
    db.session.commit()

    db.session.add(Post(title="Hello", body="First", user_id=ada.id))
    db.session.commit()

    User.query.all()
    User.query.filter_by(email="ada@example.com").first()
    db.session.get(User, 1)
    Post.query.filter(Post.title.contains("Hel")).order_by(Post.created.desc()).limit(10).all()

    ada.name = "Ada Lovelace"       # update by changing attributes
    db.session.commit()

    db.session.delete(ada)
    db.session.commit()

Changes are staged in the session and only written when you call commit(). If something fails, call db.session.rollback().

@app.route("/posts")
def posts():
    posts = Post.query.order_by(Post.created.desc()).all()
    return jsonify([{"id": p.id, "title": p.title, "author": p.author.name} for p in posts])

@app.route("/posts/<int:post_id>")
def post_detail(post_id):
    post = db.get_or_404(Post, post_id)
    return jsonify(id=post.id, title=post.title)

Migrations

When models change, create_all will not alter existing tables. Flask-Migrate (Alembic) records schema changes as versioned scripts.

flask db init
flask db migrate -m "add posts"
flask db upgrade

N+1 queries

Looping over posts and reading p.author triggers one query per post. Load related rows in one go with Post.query.options(joinedload(Post.author)).

Try it yourself

Add a Comment model belonging to a Post, then a route that returns a post with its comments as JSON.

Show solution

class Comment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    text = db.Column(db.Text, nullable=False)
    post_id = db.Column(db.Integer, db.ForeignKey("post.id"), nullable=False)
    post = db.relationship("Post", backref="comments")

@app.route("/posts/<int:pid>/full")
def full(pid):
    p = db.get_or_404(Post, pid)
    return jsonify(title=p.title, comments=[c.text for c in p.comments])