Usecase: Speedup SQLAlchemy queries

README-sqlalchemy-query-relationsihps.md

Test SQLAlchemy application

Slides: https://docs.google.com/presentation/d/1SMtBILSrqt9SWK5BnEKe4ivGTHItrVjgCEB0AUhGeyM

Setup

$ python -m venv venv
$ venv/bin/pip install sqlalchemy colorama

Interactive Shell

# Install ipython
$ venv/bin/pip install ipython

# Start a terminal
$ venv/bin/ipython

Seed some data

from app import db
from models import Author, Book, Series

tolkien = Author(name='J R R Tolkien')
lotr = Series(name='Lord of the Rings', books=[
    Book(name='Fellowship of the Ring', authors=[tolkien]),
    Book(name='Two Towers', authors=[tolkien]),
    Book(name='Return of the King', authors=[tolkien]),
])
db.add(lotr)

adams = Author(name='Douglas Adams')
h2g2 = Series(name='Hitchhiker\'s Guide to the Galaxy', books=[
    Book(name='Hitchhiker\'s Guide to the Galaxy', authors=[adams]),
    Book(name='Restaurant at the End of the Universe', authors=[adams]),
    Book(name='Life, Universe and Everything', authors=[adams]),
    Book(name='So Long, and Thanks for All the Fish', authors=[adams]),
    Book(name='Mostly Harmless', authors=[adams]),
    Book(name='And Another Thing...', authors=[adams]),
])
db.add(h2g2)

wells = Author(name='HG Wells')
time_machine = Book(name='Time Machine', authors=[wells])
db.add(time_machine)

ditko = Author(name='Steve Ditko')
stanlee = Author(name='Stan Lee')
spider_man = Book(name='Spider Man', authors=[ditko, stanlee])
db.add(spider_man)

Commands to run

Import the Models:

from app import db
from models import Author, Book, Series

Querying with sqlalchemy:

Author.query.all()

# WHERE conditions
Author.query.filter_by(name='a').all()

# Access relationships
author = Author.query.first()
author.books

# Count number of books
{a: len(a.books) for a in Author.query.all()}

# Change the loading strategy
from sqlalchemy.orm import joinedload
{a: len(a.books) for a in Author.query.options(joinedload(Author.books)).all()}

app.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import Base

# Create engine
engine = create_engine('sqlite:///db.sqlite', echo=True)

# Change logging from sqlalchemy to be grey (so it doesn't distract us)
import logging
from colorama import Fore
logger = logging.getLogger('sqlalchemy.engine.Engine')
logger.handlers[0].setFormatter(logging.Formatter(Fore.LIGHTBLACK_EX + '%(message)s' + Fore.RESET))

# Create session
session = sessionmaker()
session.configure(bind=engine)

# Create all models we need
Base.metadata.create_all(engine)

db = session()

logic.py

from sqlalchemy.orm import selectinload

from app import db
from models import Author, Book, Series


def get_all_series():
    data = db.query(Series).all()

    for series in data:
        print(f"SERIES : {series.name}:")
        for book in series.books:
            print(f" - BOOK : {book.name}")
            for author in book.authors:
                print(f"   - AUTHOR : {author.name}")


def get_all_series_opt():
    data = (
        db.query(Series)
        .options(
            selectinload(Series.books).selectinload(Book.authors),
        )
        .all()
    )

    for series in data:
        print(f"SERIES : {series.name}:")
        for book in series.books:
            print(f" - BOOK : {book.name}")
            for author in book.authors:
                print(f"   - AUTHOR : {author.name}")

models.py

from sqlalchemy import Column, String, Table, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class BaseModel(Base):
    __abstract__ = True

    def __repr__(self):
        return super().__repr__() + f" ({self.name})"


book_author = Table(
    "book_author",
    Base.metadata,
    Column("author_id", ForeignKey("author.id"), primary_key=True),
    Column("book_id", ForeignKey("book.id"), primary_key=True),
)


class Author(BaseModel):
    __tablename__ = "author"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # age = Column(Integer, nullable=True)
    # active = Column(Integer, default=1)

    books = relationship("Book", secondary=book_author, back_populates="authors", lazy='selectin')


class Book(BaseModel):
    __tablename__ = "book"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    # published_year = Column(Integer, nullable=True)
    # published_by = Column(String, nullable=True)
    # genres = Column(String, nullable=True)

    authors = relationship("Author", secondary=book_author, back_populates="books", lazy='selectin')

    series_id = Column(Integer, ForeignKey("series.id"), nullable=True)
    series = relationship("Series", back_populates="books", lazy='selectin')


class Series(BaseModel):
    __tablename__ = "series"
    id = Column(Integer, primary_key=True)
    name = Column(String)

    books = relationship("Book", back_populates="series", lazy='selectin')

    # @property
    # def length(self):
    #     return len(self.books)

schema.py

from marshmallow import Schema, fields
from sqlalchemy import inspect


class BaseSchema(Schema):
    pass


class AuthorSchema(BaseSchema):
    id = fields.Int()
    name = fields.Str()


class BookSchema(BaseSchema):
    id = fields.Int()
    name = fields.Str()

    authors = fields.Nested(AuthorSchema)


class SeriesSchema(BaseSchema):
    id = fields.Int()
    name = fields.Str()

    books = fields.Nested(BookSchema)


def get_paths(cls, model):
    relations = []
    for key, field in cls._declared_fields.items():
        data_key = field.attribute or field.name or key
        print("Got column:", data_key)
        if data_key in inspect(model).relationships:
            rel = getattr(model, data_key)
            relations.append(rel)
            print("Got relationship:", rel.key)

    return relations