Flask-SQLAlchemy

python -m venv .venv
source .venv/bin/activate
pip install "Flask-SQLAlchemy>=3.1" psycopg[binary]

import enum
import secrets
from datetime import datetime

import sqlalchemy as sa
from flask import Flask, abort, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import ForeignKey, func
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    WriteOnlyMapped,
    mapped_column,
    relationship,
)
from werkzeug.security import generate_password_hash


class Base(DeclarativeBase):
    """Common base class; metadata and naming conventions live here."""

    metadata = sa.MetaData(
        naming_convention={
            "ix": "ix_%(column_0_label)s",
            "uq": "uq_%(table_name)s_%(column_0_name)s",
            "ck": "ck_%(table_name)s_%(constraint_name)s",
            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
            "pk": "pk_%(table_name)s",
        }
    )


db = SQLAlchemy(model_class=Base)
print(db.Model.__name__, db.Model.metadata is Base.metadata)

def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_mapping(
        SQLALCHEMY_DATABASE_URI="sqlite://",
        SECRET_KEY=secrets.token_hex(),
    )
    if test_config:
        app.config.update(test_config)
    db.init_app(app)
    return app


app = create_app({"TESTING": True})
with app.app_context():
    print(db.engine.url, list(db.engines))

import os

from sqlalchemy.engine import URL

db_url = URL.create(
    "postgresql+psycopg",
    username=os.environ.get("BOOKSHELF_DB_USER", "bookshelf"),
    password=os.environ.get("BOOKSHELF_DB_PASSWORD", ""),
    host=os.environ.get("BOOKSHELF_DB_HOST", "localhost"),
    port=5432,
    database=os.environ.get("BOOKSHELF_DB_NAME", "bookshelf"),
)
print(db_url.render_as_string(hide_password=True))

import tempfile
from pathlib import Path

tmp = Path(tempfile.mkdtemp())
pool_app = Flask(__name__, instance_path=str(tmp))
pool_app.config.update(
    SQLALCHEMY_DATABASE_URI="sqlite:///pool-demo.db",
    SQLALCHEMY_ENGINE_OPTIONS={
        "pool_size": 5,
        "max_overflow": 10,
        "pool_timeout": 10,
        "pool_recycle": 1800,
        "pool_pre_ping": True,
    },
)
pool_db = SQLAlchemy(model_class=Base)
pool_db.init_app(pool_app)
with pool_app.app_context():
    print(pool_db.engine.url.database)
    print(type(pool_db.engine.pool).__name__, pool_db.engine.pool.size())

class Format(enum.Enum):
    PAPERBACK = "paperback"
    HARDCOVER = "hardcover"
    EBOOK = "ebook"


book_tag = db.Table(
    "book_tag",
    sa.Column("book_id", ForeignKey("book.id", ondelete="CASCADE"), primary_key=True),
    sa.Column("tag_id", ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True),
)


class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(sa.String(254), unique=True)
    password_hash: Mapped[str] = mapped_column(sa.String(256))
    is_admin: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    reviews: WriteOnlyMapped["Review"] = relationship(
        back_populates="user", passive_deletes=True
    )

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


class Author(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(120), index=True)
    bio: Mapped[str | None] = mapped_column(sa.Text)

    books: Mapped[list["Book"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",
        order_by="Book.title",
    )


class Book(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    isbn: Mapped[str] = mapped_column(sa.String(17), unique=True)
    title: Mapped[str] = mapped_column(sa.String(200), index=True)
    published_year: Mapped[int | None]
    price: Mapped[float] = mapped_column(sa.Numeric(8, 2, asdecimal=False), default=0)
    format: Mapped[Format] = mapped_column(default=Format.PAPERBACK)
    cover_path: Mapped[str | None] = mapped_column(sa.String(255))
    author_id: Mapped[int] = mapped_column(ForeignKey("author.id", ondelete="CASCADE"))

    author: Mapped[Author] = relationship(back_populates="books")
    reviews: WriteOnlyMapped["Review"] = relationship(
        back_populates="book",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    tags: Mapped[list["Tag"]] = relationship(
        secondary=book_tag, back_populates="books"
    )

    def __repr__(self) -> str:
        return f"<Book {self.isbn}>"


class Tag(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(40), unique=True)
    books: Mapped[list[Book]] = relationship(secondary=book_tag, back_populates="tags")


class Review(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    rating: Mapped[int]
    body: Mapped[str] = mapped_column(sa.Text, default="")
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    book_id: Mapped[int] = mapped_column(ForeignKey("book.id", ondelete="CASCADE"))
    user_id: Mapped[int] = mapped_column(ForeignKey("user.id", ondelete="CASCADE"))

    book: Mapped[Book] = relationship(back_populates="reviews")
    user: Mapped[User] = relationship(back_populates="reviews")

    __table_args__ = (
        sa.CheckConstraint("rating BETWEEN 1 AND 5", name="rating_range"),
    )


print([t.name for t in db.metadata.sorted_tables])
print(User.__tablename__, Book.__tablename__, [c.name for c in Review.__table__.constraints if c.name][:2])

def seeded_app():
    """A fresh application with the Bookshelf tables and sample rows."""
    new_app = create_app({"TESTING": True})
    with new_app.app_context():
        db.create_all()
        password = generate_password_hash(secrets.token_urlsafe(16))
        ada = User(email="ada@example.com", password_hash=password, is_admin=True)
        martin = Author(name="Robert C. Martin", bio="Software craftsman")
        clean_code = Book(
            isbn="978-0132350884",
            title="Clean Code",
            published_year=2008,
            price=33.5,
            format=Format.PAPERBACK,
            author=martin,
        )
        clean_arch = Book(
            isbn="978-0134494166", title="Clean Architecture", price=29.99, author=martin
        )
        clean_code.tags = [Tag(name="craft"), Tag(name="design")]
        clean_arch.tags = [clean_code.tags[1]]
        db.session.add_all([ada, martin])
        db.session.flush()
        clean_code.reviews.add(Review(rating=5, body="Essential.", user=ada))
        db.session.commit()
    return new_app


app = seeded_app()
with app.app_context():
    martin = db.session.scalar(db.select(Author))
    clean_arch, clean_code = martin.books
    print([b.title for b in martin.books])
    print([t.name for t in clean_arch.tags], [b.isbn for b in clean_arch.tags[0].books])
    print(db.session.scalar(sa.select(func.count()).select_from(clean_code.reviews.select().subquery())))

import click


cli_app = create_app({"TESTING": True})


@cli_app.cli.command("init-db")
@click.option("--drop", is_flag=True, help="Drop all tables first (development only).")
def init_db_command(drop):
    """Create the Bookshelf tables."""
    if drop:
        db.drop_all()
    db.create_all()
    click.echo("Initialized the database.")


runner = cli_app.test_cli_runner()
result = runner.invoke(args=["init-db", "--drop"])
print(result.output.strip())

with app.app_context():
    print(sa.inspect(db.engine).get_table_names())

with app.app_context():
    # all books ordered by title: scalars() unwraps the one-column rows
    books = db.session.scalars(db.select(Book).order_by(Book.title)).all()
    print([b.title for b in books])

    # primary-key lookup: uses the session's identity map first
    book = db.session.get(Book, 1)
    print(book, db.session.get(Book, 999))

    # filter_by uses keyword equality, where() takes SQL expressions
    one = db.session.scalar(db.select(Book).filter_by(isbn="978-0134494166"))
    cheap = db.session.scalars(db.select(Book).where(Book.price < 30)).all()
    print(one.title, [b.title for b in cheap])

    # join, aggregate and group
    stmt = (
        db.select(Author.name, func.count(Book.id).label("n"))
        .join(Book)
        .group_by(Author.id)
        .having(func.count(Book.id) > 1)
    )
    print([tuple(r) for r in db.session.execute(stmt)])

    # exists / count
    total = db.session.scalar(db.select(func.count()).select_from(Book))
    has_reviews = db.session.scalar(
        db.select(sa.exists().where(Review.book_id == Book.id)).select_from(Book).limit(1)
    )
    print(total, has_reviews)

or404_app = seeded_app()


@or404_app.get("/books/<int:book_id>")
def book_detail(book_id: int):
    book = db.get_or_404(Book, book_id, description="No such book.")
    return jsonify(id=book.id, title=book.title, author=book.author.name)


@or404_app.get("/authors/<name>")
def author_by_name(name: str):
    author = db.one_or_404(db.select(Author).filter_by(name=name))
    return jsonify(id=author.id, books=[b.title for b in author.books])


client = or404_app.test_client()
print(client.get("/books/1").get_json())
print(client.get("/books/999").status_code, client.get("/authors/Nobody").status_code)

page_app = seeded_app()


@page_app.get("/books")
def list_books():
    page = db.paginate(db.select(Book).order_by(Book.title), max_per_page=50)
    return jsonify(
        items=[{"id": b.id, "title": b.title} for b in page],
        page=page.page,
        pages=page.pages,
        total=page.total,
        has_next=page.has_next,
        next_num=page.next_num,
    )


page_client = page_app.test_client()
print(page_client.get("/books?per_page=1&page=2").get_json())
print(page_client.get("/books?page=9").status_code, page_client.get("/books?page=0").status_code)

with page_app.test_request_context("/books?page=7"):
    sample = db.paginate(db.select(Book), per_page=1, error_out=False, count=False)
    sample.total = 20  # supplied by the caller because count=False
    print(list(sample.iter_pages(left_edge=1, left_current=1, right_current=2, right_edge=1)))

{% for p in page.iter_pages() %}
  {% if p %}
    <a href="{{ url_for('list_books', page=p) }}"{% if p == page.page %} aria-current="page"{% endif %}>{{ p }}</a>
  {% else %}
    <span>&hellip;</span>
  {% endif %}
{% endfor %}

from sqlalchemy.exc import IntegrityError


tx_app = seeded_app()


@tx_app.post("/books")
def create_book():
    data = request.get_json()
    author = db.get_or_404(Author, data["author_id"])
    book = Book(isbn=data["isbn"], title=data["title"], author=author)
    db.session.add(book)
    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        abort(409, description="A book with this ISBN already exists.")
    return jsonify(id=book.id, title=book.title), 201


payload = {"isbn": "978-0137081073", "title": "The Clean Coder", "author_id": 1}
tx_client = tx_app.test_client()
print(tx_client.post("/books", json=payload).status_code)
print(tx_client.post("/books", json=payload).status_code)

class ReportBase(DeclarativeBase):
    pass


report_db = SQLAlchemy(model_class=ReportBase)


class Subscriber(report_db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(sa.String(254))


class PageView(report_db.Model):
    __bind_key__ = "analytics"
    id: Mapped[int] = mapped_column(primary_key=True)
    path: Mapped[str] = mapped_column(sa.String(255))


multi_app = Flask(__name__)
multi_app.config.update(
    SQLALCHEMY_DATABASE_URI="sqlite://",
    SQLALCHEMY_BINDS={"analytics": {"url": "sqlite://", "echo": False}},
)
report_db.init_app(multi_app)
with multi_app.app_context():
    report_db.create_all()  # every bind
    report_db.session.add_all([Subscriber(email="reader@example.com"), PageView(path="/books")])
    report_db.session.commit()
    print(sorted(str(k) for k in report_db.engines), report_db.engines["analytics"].url)
    print(report_db.session.scalar(sa.select(func.count()).select_from(PageView)))
    print([t.name for t in report_db.metadatas["analytics"].sorted_tables], report_db.metadata.tables.keys())

from flask_sqlalchemy.record_queries import get_recorded_queries

debug_app = create_app({"SQLALCHEMY_RECORD_QUERIES": True, "SQLALCHEMY_ECHO": False})


@debug_app.get("/slow-report")
def slow_report():
    db.session.scalars(db.select(Book)).all()
    stats = [
        {"sql": q.statement.splitlines()[0][:40], "ms": round(q.duration * 1000, 2)}
        for q in get_recorded_queries()
    ]
    return jsonify(count=len(stats), first=stats[0]["sql"])


with debug_app.app_context():
    db.create_all()
print(debug_app.test_client().get("/slow-report").get_json())

def make_test_app():
    test_app = create_app({"TESTING": True, "SQLALCHEMY_DATABASE_URI": "sqlite://"})
    with test_app.app_context():
        db.create_all()
    test_app.add_url_rule("/books", view_func=create_book, methods=["POST"])
    return test_app


def test_duplicate_isbn_returns_409():
    test_app = make_test_app()
    with test_app.app_context():
        db.session.add(Author(name="A"))
        db.session.commit()
    c = test_app.test_client()
    body = {"isbn": "123", "title": "T", "author_id": 1}
    assert c.post("/books", json=body).status_code == 201
    assert c.post("/books", json=body).status_code == 409


test_duplicate_isbn_returns_409()
print("test passed")

# norun (conftest.py; verified with pytest on a minimal bookshelf package)
import pytest

from bookshelf import create_app
from bookshelf.extensions import db


@pytest.fixture()
def app():
    app = create_app({"TESTING": True, "SQLALCHEMY_DATABASE_URI": "sqlite://"})
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()


@pytest.fixture()
def client(app):
    return app.test_client()

from flask_sqlalchemy_lite import SQLAlchemy as LiteSQLAlchemy


class LiteBase(DeclarativeBase):
    pass


class Shelf(LiteBase):
    __tablename__ = "shelf"
    id: Mapped[int] = mapped_column(primary_key=True)
    label: Mapped[str] = mapped_column(sa.String(40))


lite = LiteSQLAlchemy()
lite_app = Flask(__name__)
lite_app.config["SQLALCHEMY_ENGINES"] = {"default": "sqlite://"}
lite.init_app(lite_app)

with lite_app.app_context():
    LiteBase.metadata.create_all(lite.engine)
    lite.session.add(Shelf(label="Classics"))
    lite.session.commit()
    print(lite.session.scalars(sa.select(Shelf)).all()[0].label)
    print(lite.get_or_abort(Shelf, 1).label)

python -c 'import secrets; print("BOOKSHELF_DB_PASSWORD=" + secrets.token_urlsafe(24))' > .env
echo ".env" >> .gitignore

services:
  db:
    image: postgres:17
    environment:
      POSTGRES_USER: bookshelf
      POSTGRES_DB: bookshelf
      POSTGRES_PASSWORD: ${BOOKSHELF_DB_PASSWORD:?set it in .env}
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - bookshelf-pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U bookshelf"]
      interval: 5s
      retries: 10

volumes:
  bookshelf-pgdata: