11.3 — Flask with SQLAlchemy

Theory 55 min Advanced

Why SQLAlchemy?

SQLAlchemy is the most popular Python ORM (Object-Relational Mapper). It lets you work with databases using Python classes and objects instead of writing raw SQL.

Flask-SQLAlchemy integrates SQLAlchemy with Flask, handling connection management, session lifecycle, and configuration.

┌──────────────────────────────────────────────────────────────┐
│              ORM Layer                                        │
│                                                              │
│  Python Object         ORM               SQL Database        │
│  ──────────────────────────────────────────────────────────  │
│  user = User(...)  ──▶ INSERT INTO users VALUES (...)        │
│  User.query.all()  ──▶ SELECT * FROM users                   │
│  user.name = "X"   ──▶ UPDATE users SET name='X' WHERE …    │
│  db.session.delete(u) ▶ DELETE FROM users WHERE id=…         │
└──────────────────────────────────────────────────────────────┘

---

Setup

pip install flask-sqlalchemy flask-migrate

Project structure

project/
├── app.py               ← application factory
├── models.py            ← SQLAlchemy models
├── routes/
│   ├── users.py
│   └── posts.py
├── migrations/          ← auto-generated by Flask-Migrate
└── instance/
    └── app.db           ← SQLite file (dev only)

---

Configuration

# app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db      = SQLAlchemy()    # shared db instance (no app yet)
migrate = Migrate()

def create_app():
    app = Flask(__name__)
    app.config["SECRET_KEY"]             = "dev-secret-change-me"
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

    db.init_app(app)
    migrate.init_app(app, db)

    from routes.users import bp as users_bp
    app.register_blueprint(users_bp)

    return app

SQLALCHEMY_DATABASE_URI formats:

• SQLite: sqlite:///app.db (relative) or sqlite:////abs/path/app.db
• PostgreSQL: postgresql://user:pass@localhost/dbname
• MySQL: mysql+pymysql://user:pass@localhost/dbname

---

Models

# models.py
from datetime import datetime, timezone
from app import db

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

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

    # One-to-many: one User has many Posts
    posts = db.relationship("Post", back_populates="author", cascade="all, delete-orphan")

    def to_dict(self):
        return {
            "id":         self.id,
            "username":   self.username,
            "email":      self.email,
            "created_at": self.created_at.isoformat(),
        }

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


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

    id         = db.Column(db.Integer, primary_key=True)
    title      = db.Column(db.String(200), nullable=False)
    body       = db.Column(db.Text, nullable=False)
    published  = db.Column(db.Boolean, default=False, nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))

    author_id  = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    author     = db.relationship("User", back_populates="posts")

    def to_dict(self):
        return {
            "id":        self.id,
            "title":     self.title,
            "published": self.published,
            "author":    self.author.username,
        }

---

Database Migrations

Flask-Migrate (uses Alembic under the hood) tracks schema changes without losing data:

# Initialize migrations directory (first time only)
flask db init

# Auto-detect model changes and create a migration script
flask db migrate -m "add users and posts tables"

# Apply the migration to the database
flask db upgrade

# Roll back one migration
flask db downgrade

---

CRUD Routes

# routes/users.py
from flask import Blueprint, jsonify, request
from app import db
from models import User

bp = Blueprint("users", __name__, url_prefix="/api/users")

# ── List ──────────────────────────────────────────────────────
@bp.route("", methods=["GET"])
def list_users():
    page     = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 20, type=int)
    paginated = User.query.order_by(User.created_at.desc()).paginate(
        page=page, per_page=per_page, error_out=False
    )
    return jsonify({
        "users":   [u.to_dict() for u in paginated.items],
        "total":   paginated.total,
        "pages":   paginated.pages,
        "page":    paginated.page,
    })

# ── Get one ───────────────────────────────────────────────────
@bp.route("/<int:user_id>", methods=["GET"])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

# ── Create ────────────────────────────────────────────────────
@bp.route("", methods=["POST"])
def create_user():
    data = request.get_json(silent=True) or {}

    if not data.get("username") or not data.get("email"):
        return jsonify({"error": "username and email are required"}), 422

    # Check uniqueness
    if User.query.filter_by(username=data["username"]).first():
        return jsonify({"error": "Username already taken"}), 409

    user = User(username=data["username"], email=data["email"])
    db.session.add(user)
    db.session.commit()

    return jsonify(user.to_dict()), 201

# ── Update ────────────────────────────────────────────────────
@bp.route("/<int:user_id>", methods=["PATCH"])
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.get_json(silent=True) or {}

    if "email" in data:
        user.email = data["email"]
    if "username" in data:
        user.username = data["username"]

    db.session.commit()
    return jsonify(user.to_dict())

# ── Delete ────────────────────────────────────────────────────
@bp.route("/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return "", 204

---

Querying

# Basic queries
all_users      = User.query.all()
first_user     = User.query.first()
user_by_id     = User.query.get(1)            # deprecated in SQLAlchemy 2.0
user_by_id     = db.session.get(User, 1)      # SQLAlchemy 2.0 style

# Filtering
admins         = User.query.filter_by(is_admin=True).all()
alice          = User.query.filter(User.username == "alice").first()
recent         = User.query.filter(User.created_at > cutoff).all()

# Ordering and limiting
top5           = User.query.order_by(User.created_at.desc()).limit(5).all()

# Counting
total          = User.query.count()
active_count   = User.query.filter_by(active=True).count()

# Joins
posts_with_author = db.session.query(Post).join(User).filter(User.username == "alice").all()

---

Simple Authentication with JWT

pip install flask-jwt-extended

from flask import Flask, jsonify, request
from flask_jwt_extended import (
    JWTManager, create_access_token,
    jwt_required, get_jwt_identity
)

app = Flask(__name__)
app.config["JWT_SECRET_KEY"] = "super-secret"
jwt = JWTManager(app)

# In-memory users (use a DB + hashed passwords in production)
USERS = {"alice": "password123"}

@app.route("/auth/login", methods=["POST"])
def login():
    data     = request.get_json()
    username = data.get("username")
    password = data.get("password")

    if USERS.get(username) != password:
        return jsonify({"error": "Invalid credentials"}), 401

    token = create_access_token(identity=username)
    return jsonify({"access_token": token})

@app.route("/api/me", methods=["GET"])
@jwt_required()                          # requires Bearer token
def me():
    current_user = get_jwt_identity()    # returns the identity we passed to create_access_token
    return jsonify({"user": current_user})

# Get a token
http POST localhost:5000/auth/login username=alice password=password123
# → {"access_token": "eyJ..."}

# Use the token
http GET localhost:5000/api/me Authorization:"Bearer eyJ..."

---

Deployment Checklist

# config.py
import os

class Config:
    SECRET_KEY           = os.environ.get("SECRET_KEY", "change-me")
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", "sqlite:///app.db")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    DEBUG                = False

class DevelopmentConfig(Config):
    DEBUG = True

class ProductionConfig(Config):
    pass

config = {
    "development": DevelopmentConfig,
    "production":  ProductionConfig,
}

# Production server (Gunicorn)
pip install gunicorn

# Run with 4 worker processes
gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()"

Production checklist:

Item	Status
DEBUG=False	✅ required
SECRET_KEY from env var	✅ required
HTTPS only	✅ required
Hashed passwords (bcrypt)	✅ required
DATABASE_URL from env var	✅ required
Rate limiting	✅ recommended
Error monitoring (Sentry)	✅ recommended

---

Key Vocabulary

Term	Definition
ORM	Object-Relational Mapper — maps Python classes to DB tables
Migration	Versioned database schema change script
Session	SQLAlchemy's unit-of-work — tracks changes and flushes them as a transaction
Relationship	ORM link between two models (one-to-many, many-to-many)
JWT	JSON Web Token — stateless authentication token
Gunicorn	Production WSGI server for Flask/Django apps
Application factory	create_app() pattern for testable, configurable Flask apps

---

Summary

• Flask-SQLAlchemy maps Python classes to database tables
• db.Column(type, ...) defines table columns with constraints
• db.relationship(...) defines ORM-level links between models
• flask db migrate && flask db upgrade applies schema changes safely
• User.query.filter_by(...).all() and .paginate(...) for flexible querying
• flask-jwt-extended handles token-based authentication
• Deploy with Gunicorn, use environment variables for secrets, set DEBUG=False

Congratulations — you've completed the Python course from fundamentals to production web APIs!