Hands-on lab — Day 93: ORMs and SQLAlchemy

"""counting.py — record every statement the engine sends to the database.

This is the instrument the whole lab is built around. The ORM's promise is that
you write Python and it writes SQL; the only way to hold it to that promise is
to look at the SQL. SQLAlchemy exposes a `before_cursor_execute` event on the
Engine, which fires once per statement actually sent to the DBAPI cursor, and
that is exactly the granularity a query count wants.

Why count statements rather than time them: a timing assertion is a flake
waiting for a slow machine, and it tells you nothing about the cause. A count
is deterministic, it is the same number on every machine, and it names the
defect directly — "this loop issued fifty-one queries" is a bug report, while
"this loop took 240 milliseconds" is a mood.
"""

from __future__ import annotations

from sqlalchemy import event
from sqlalchemy.engine import Engine


def normalise(statement: str) -> str:
    """Collapse SQLAlchemy's multi-line SQL into one comparable line."""
    return " ".join(statement.split())


class QueryCounter:
    """Context manager recording every statement an Engine emits.

    Usage:

        with QueryCounter(engine) as counted:
            ...do ORM work...
        print(len(counted), counted.selects())
    """

    def __init__(self, engine: Engine, seed: bool = False) -> None:
        self.engine = engine
        self.statements: list[str] = []
        self.batched: list[bool] = []
        self.parameter_sets: list[int] = []
        self._seed = seed

    def _record(self, conn, cursor, statement, parameters, context, executemany):
        self.statements.append(normalise(statement))
        self.batched.append(bool(executemany))
        # How many rows this one cursor execution carries. An executemany hands
        # the driver a sequence of parameter tuples; a normal execute hands it
        # one. Counting both numbers is what stops "one statement" from being
        # mistaken for "one row".
        if executemany:
            try:
                self.parameter_sets.append(len(parameters))
            except TypeError:
                self.parameter_sets.append(1)
        else:
            self.parameter_sets.append(1)

    def __enter__(self) -> QueryCounter:
        event.listen(self.engine, "before_cursor_execute", self._record)
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        event.remove(self.engine, "before_cursor_execute", self._record)
        return False

    def __len__(self) -> int:
        """Cursor executions — the number of round trips to the driver."""
        return len(self.statements)

    def rows_sent(self) -> int:
        """Parameter sets across every execution: how many rows were carried."""
        return sum(self.parameter_sets)

    def executemany_count(self) -> int:
        """How many of the executions were batched executemany calls."""
        return sum(1 for flag in self.batched if flag)

    def starting_with(self, keyword: str) -> list[str]:
        upper = keyword.upper()
        return [s for s in self.statements if s.upper().startswith(upper)]

    def selects(self) -> list[str]:
        return self.starting_with("SELECT")

    def inserts(self) -> list[str]:
        return self.starting_with("INSERT")

    def updates(self) -> list[str]:
        return self.starting_with("UPDATE")

    def report(self, indent: str = "    ") -> str:
        if not self.statements:
            return f"{indent}(no statements)"
        lines = []
        for number, statement in enumerate(self.statements, start=1):
            shown = statement if len(statement) <= 110 else statement[:107] + "..."
            lines.append(f"{indent}{number:>2}. {shown}")
        return "\n".join(lines)

"""demo_bulk.py — where the ORM stops being the right tool, measured honestly.

Run it:  python3 examples/demo_bulk.py

An ORM's job is to track individual objects through a unit of work. That is
the wrong shape for "change five hundred rows", and this script measures by
how much — but the measurement is more interesting than the folklore, so read
the numbers rather than the slogan.

Two numbers are recorded for every approach, and keeping them apart is the
whole lesson of this file:

  * **cursor executions** — how many times a statement was handed to the
    driver. This is the round-trip count, and it is what people mean when they
    say "number of queries".
  * **parameter sets** — how many rows those executions carried. One
    `executemany` call is ONE execution carrying five hundred rows.

Conflate the two and you will reach a conclusion the machine does not support.
"""

from __future__ import annotations

from sqlalchemy import insert, select, update
from sqlalchemy.orm import Session

from counting import QueryCounter
from library import build_engine
from models import Loan

ROWS = 500


def rule(label: str) -> None:
    print()
    print(label)
    print("-" * len(label))


def new_loans(start_id: int, count: int = ROWS) -> list[dict]:
    return [
        {
            "id": start_id + offset,
            "book_id": (offset % 8) + 1,
            "member_id": (offset % 6) + 1,
            "borrowed_on": "2026-08-01",
            "due_on": "2026-08-22",
            "returned": False,
        }
        for offset in range(count)
    ]


def summarise(counted: QueryCounter) -> str:
    return (
        f"{len(counted)} cursor execution(s), "
        f"{counted.rows_sent()} parameter set(s), "
        f"{counted.executemany_count()} of them batched"
    )


def main() -> None:
    rule(f"1. Inserting {ROWS} rows the naive way: add() and flush() in the loop")
    engine = build_engine()
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            for row in new_loans(1001):
                session.add(Loan(**row))
                session.flush()
        session.rollback()
    naive = len(counted)
    print(summarise(counted))
    print(f"    {counted.statements[0]}")
    print(f"    ... and {naive - 1} more identical statements")
    print("  One round trip per row. This is the mistake, and it is a mistake")
    print("  about WHERE the flush goes, not about the ORM.")
    engine.dispose()

    rule(f"2. Inserting {ROWS} rows with add_all() and ONE flush")
    engine = build_engine()
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            session.add_all([Loan(**row) for row in new_loans(1001)])
            session.flush()
        session.rollback()
    batched = len(counted)
    batched_rows = counted.rows_sent()
    print(summarise(counted))
    print("  The unit of work sorted the pending objects by table and batched")
    print("  them into a single executemany. Still the ORM: every object is")
    print("  tracked, every identity registered, every default applied.")
    engine.dispose()

    rule(f"3. Inserting {ROWS} rows through Core")
    engine = build_engine()
    with engine.begin() as connection:
        with QueryCounter(engine) as counted:
            connection.execute(insert(Loan), new_loans(1001))
    core = len(counted)
    core_rows = counted.rows_sent()
    print(summarise(counted))
    print(f"    {counted.statements[0][:96]}...")
    engine.dispose()

    rule("4. What the insert numbers actually say")
    print(f"    add() + flush() per row   {naive:>4} execution(s)   {ROWS:>4} row(s)")
    print(f"    add_all() + one flush     {batched:>4} execution(s)   {batched_rows:>4} row(s)")
    print(f"    Core insert(), one call   {core:>4} execution(s)   {core_rows:>4} row(s)")
    print()
    print("  Read that carefully, because it contradicts the usual advice.")
    print("  On this version, batched ORM inserts and Core inserts issue the")
    print("  SAME number of cursor executions. The dramatic gap is between the")
    print("  naive loop and everything else — it is 500 against 1.")
    print()
    print("  So 'drop to Core for speed' is not what the execution count shows.")
    print("  What Core actually saves here is Python-side work the counter")
    print("  cannot see: no Loan instances are constructed, nothing enters the")
    print("  identity map, and the unit of work has no dependency graph to")
    print("  sort. That is real, and it is a memory and CPU argument rather")
    print("  than a round-trip argument. Measure it before you claim it.")

    rule("5. Updating every open loan — where Core genuinely wins")
    engine = build_engine()
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            loaded = 0
            for loan in session.scalars(select(Loan).where(Loan.returned.is_(False))):
                loan.returned = True
                loaded += 1
            session.flush()
        session.rollback()
    orm_update = len(counted)
    print(f"ORM, object by object : {summarise(counted)}")
    print(f"                        {loaded} Loan objects built in memory")
    print(counted.report())
    engine.dispose()

    engine = build_engine()
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            result = session.execute(
                update(Loan).where(Loan.returned.is_(False)).values(returned=True)
            )
            changed = result.rowcount
        session.rollback()
    core_update = len(counted)
    print(f"Core UPDATE           : {summarise(counted)}")
    print(f"                        0 Loan objects built, {changed} rows changed")
    print(counted.report())
    print("  Note the shape of the difference. The ORM had to SELECT the rows")
    print("  first, because it changes objects and it has no objects until it")
    print("  loads them. Core changes rows, so it never reads them.")
    engine.dispose()

    rule("6. That difference grows with the row count; the insert one does not")
    engine = build_engine()
    with Session(engine) as session:
        session.execute(insert(Loan), new_loans(2001, 1000))
        session.commit()
        with QueryCounter(engine) as counted:
            loaded = 0
            for loan in session.scalars(select(Loan).where(Loan.returned.is_(False))):
                loan.returned = True
                loaded += 1
            session.flush()
        session.rollback()
    print(f"ORM with ~1000 more open loans : {summarise(counted)}")
    print(f"                                 {loaded} Loan objects built in memory")
    engine.dispose()

    engine = build_engine()
    with Session(engine) as session:
        session.execute(insert(Loan), new_loans(2001, 1000))
        session.commit()
        with QueryCounter(engine) as counted:
            changed = session.execute(
                update(Loan).where(Loan.returned.is_(False)).values(returned=True)
            ).rowcount
        session.rollback()
    print(f"Core with the same rows        : {summarise(counted)}")
    print(f"                                 0 Loan objects built, {changed} rows changed")
    print("  The execution counts barely move. The object count moves by a")
    print("  thousand. THAT is the bulk-operation argument, stated in the")
    print("  units it is actually true in.")
    engine.dispose()

    rule("7. The honest summary")
    print("  * Never flush inside a loop. That is the only order-of-magnitude")
    print("    round-trip win available here, and it is free.")
    print("  * A batched ORM insert costs the same round trips as Core. Choose")
    print("    Core for it when you do not want the objects, not for the count.")
    print("  * A bulk UPDATE or DELETE is different: the ORM must load rows to")
    print("    change them and Core does not, so Core avoids work that scales")
    print("    with the number of matching rows.")
    print("  * The price of Core is that no Python-level default, validator or")
    print("    event of yours runs, because no object was ever created. That is")
    print("    a design decision, not an optimisation.")


if __name__ == "__main__":
    main()

"""demo_n_plus_one.py — the ORM's most expensive habit, counted and then fixed.

Run it:  python3 examples/demo_n_plus_one.py

The N+1 problem is not a bug in SQLAlchemy. It is the direct consequence of a
relationship attribute being a query in disguise: `member.loans` looks like a
list, so people loop over members and touch it, and each touch is a round trip.
Nothing in the Python source hints at the cost, which is why you count.
"""

from __future__ import annotations

from sqlalchemy import select
from sqlalchemy.exc import InvalidRequestError
from sqlalchemy.orm import Session, joinedload, selectinload

from counting import QueryCounter
from library import build_engine
from models import Book, Loan, Member, Tag


def rule(label: str) -> None:
    print()
    print(label)
    print("-" * len(label))


def main() -> None:
    engine = build_engine()

    rule("1. The innocent-looking loop")
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            members = session.scalars(select(Member).order_by(Member.id)).all()
            total = sum(len(member.loans) for member in members)
    print(f"members: {len(members)}   loans reached: {total}")
    print(f"statements emitted: {len(counted)}")
    print(counted.report())
    print(f"  That is 1 + {len(members)}. The 1 is the members query; the {len(members)} are")
    print("  one lazy load per member, issued the first time .loans is touched.")
    n_plus_one = len(counted)

    rule("2. Fixed with selectinload — two statements, always")
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            members = session.scalars(
                select(Member).options(selectinload(Member.loans)).order_by(Member.id)
            ).all()
            total = sum(len(member.loans) for member in members)
    print(f"members: {len(members)}   loans reached: {total}")
    print(f"statements emitted: {len(counted)}")
    print(counted.report())
    selectin_count = len(counted)

    rule("3. Fixed with joinedload — one statement")
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            members = session.scalars(
                select(Member).options(joinedload(Member.loans)).order_by(Member.id)
            ).unique().all()
            total = sum(len(member.loans) for member in members)
    print(f"members: {len(members)}   loans reached: {total}")
    print(f"statements emitted: {len(counted)}")
    print(counted.report())
    joined_count = len(counted)

    rule("4. The scoreboard")
    print(f"    lazy (default)   {n_plus_one:>2} statements   <- 1 + N")
    print(f"    selectinload     {selectin_count:>2} statements   <- 1 + 1, whatever N is")
    print(f"    joinedload       {joined_count:>2} statement    <- 1, but wider rows")

    rule("5. Why joinedload is not simply the winner")
    statement = select(Member).options(joinedload(Member.loans)).order_by(Member.id)
    with Session(engine) as session:
        try:
            session.scalars(statement).all()
        except InvalidRequestError as error:
            print("forgetting .unique() raises, and the message says why:")
            print(f"    {str(error).splitlines()[0]}")
        raw_rows = session.connection().exec_driver_sql(
            str(statement.compile(engine))
        ).fetchall()
        distinct = session.scalars(statement).unique().all()
    print(f"rows the JOIN actually returned : {len(raw_rows)}")
    print(f"distinct Member objects built   : {len(distinct)}")
    print("  Every member's columns are repeated once per loan. With a wide parent")
    print("  row and a large collection that duplication is the cost, and it is paid")
    print("  in bytes over the wire. `.unique()` is mandatory on a joinedload of a")
    print("  collection precisely because the driver really does return those rows.")
    print()
    print("  selectinload sends a second SELECT with an IN clause instead: no")
    print("  duplication, no join, but one extra round trip. Choose joinedload for")
    print("  many-to-one and small collections; selectinload for one-to-many.")

    rule("6. It compounds — two levels of laziness")
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            loans = session.scalars(select(Loan).order_by(Loan.id)).all()
            titles = {loan.book.title for loan in loans}
    print(f"loans: {len(loans)}   distinct titles: {len(titles)}")
    print(f"statements emitted: {len(counted)}")
    print("  Not 1 + 24, because the identity map answers the second request for a")
    print("  book already loaded. The count is 1 + the number of DISTINCT books.")

    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            loans = session.scalars(
                select(Loan).options(joinedload(Loan.book)).order_by(Loan.id)
            ).all()
            titles = {loan.book.title for loan in loans}
    print(f"with joinedload(Loan.book): {len(counted)} statement, {len(titles)} titles")
    print("  This is the many-to-one case, and joinedload is the right tool for it:")
    print("  no row multiplication, because each loan has exactly one book.")

    rule("7. A many-to-many, which is where the count really bites")
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            books = session.scalars(select(Book).order_by(Book.id)).all()
            pairs = sum(len(book.tags) for book in books)
    lazy_many = len(counted)
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            books = session.scalars(
                select(Book).options(selectinload(Book.tags)).order_by(Book.id)
            ).all()
            pairs = sum(len(book.tags) for book in books)
    print(f"books: {len(books)}   book-tag pairs: {pairs}")
    print(f"    lazy         {lazy_many} statements")
    print(f"    selectinload {len(counted)} statements")
    print(counted.report())

    engine.dispose()


if __name__ == "__main__":
    main()

"""demo_sqlalchemy.py — the same four operations, in SQLAlchemy 2.0.

Run it:  python3 examples/demo_sqlalchemy.py

Everything the toy did, the real library also does — with a great deal more
care. The section numbers here deliberately match `demo_toy.py`, so you can
read the two side by side.
"""

from __future__ import annotations

from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session
from sqlalchemy.schema import CreateTable

from counting import QueryCounter
from library import build_engine
from models import Base, Book, Loan, Member, Tag


def rule(label: str) -> None:
    print()
    print(label)
    print("-" * len(label))


def main() -> None:
    import sqlalchemy

    rule("0. Versions actually in use")
    print(f"SQLAlchemy {sqlalchemy.__version__}")
    engine = create_engine("sqlite://")
    with engine.connect() as connection:
        print(f"SQLite      {connection.exec_driver_sql('select sqlite_version()').scalar_one()}")
    print(f"dialect     {engine.dialect.name}, driver {engine.dialect.driver}")
    print(f"pool        {type(engine.pool).__name__}")

    rule("1. The class declaration IS the schema")
    for table in Base.metadata.sorted_tables:
        print(str(CreateTable(table).compile(engine)).strip())

    rule("2. add() makes an object pending — no SQL yet")
    engine = build_engine()
    with Session(engine) as session:
        with QueryCounter(engine) as counted:
            member = Member(name="Grace Mensah", email="grace@library.test")
            session.add(member)
            from sqlalchemy import inspect

            state = inspect(member)
            print(f"statements emitted by add(): {len(counted)}")
            print(f"member.id                  : {member.id}   (nobody has decided it yet)")
            print(
                "state -> transient={} pending={} persistent={} detached={}".format(
                    state.transient, state.pending, state.persistent, state.detached
                )
            )

        rule("3. flush() emits the INSERT; commit() ends the transaction")
        with QueryCounter(engine) as counted:
            session.flush()
        print("after flush():")
        print(counted.report())
        print(f"member.id                  : {member.id}   (the database decided it)")
        state = inspect(member)
        print(
            "state -> transient={} pending={} persistent={} detached={}".format(
                state.transient, state.pending, state.persistent, state.detached
            )
        )
        with QueryCounter(engine) as counted:
            session.commit()
        print(f"statements emitted by commit(): {len(counted)}  (COMMIT is not a cursor execute)")

        rule("4. Rows map back into objects")
        with QueryCounter(engine) as counted:
            rows = session.scalars(select(Member).order_by(Member.id)).all()
        print(counted.report())
        for row in rows:
            print(f"    {row}")

        rule("5. The identity map: the same row is the same object")
        with QueryCounter(engine) as counted:
            first = session.get(Member, 1)
            second = session.get(Member, 1)
        print(f"first is second    : {first is second}")
        print(f"statements emitted : {len(counted)}   (already loaded in step 4)")

    rule("6. select(): filtering, ordering, joining, aggregating")
    with Session(engine) as session:
        statement = (
            select(Book.title, Book.author)
            .where(Book.copies >= 3)
            .order_by(Book.title)
        )
        print("Python:")
        print("    select(Book.title, Book.author).where(Book.copies >= 3).order_by(Book.title)")
        print("SQL:")
        print("    " + " ".join(str(statement.compile(engine)).split()))
        print("Rows:")
        for title, author in session.execute(statement):
            print(f"    {title} — {author}")

        rule("7. A join and an aggregate")
        statement = (
            select(Member.name, func.count(Loan.id).label("open_loans"))
            .join(Loan, Loan.member_id == Member.id)
            .where(Loan.returned.is_(False))
            .group_by(Member.id)
            .order_by(func.count(Loan.id).desc(), Member.name)
        )
        print("SQL:")
        print("    " + " ".join(str(statement.compile(engine)).split()))
        print("Rows:")
        for name, open_loans in session.execute(statement):
            print(f"    {name:<16} {open_loans}")

        rule("8. A many-to-many through the secondary table")
        craft = session.scalars(select(Tag).where(Tag.name == "craft")).one()
        print(f"tag: {craft}")
        for book in sorted(craft.books, key=lambda b: b.id):
            print(f"    {book}")

    engine.dispose()


if __name__ == "__main__":
    main()

"""demo_toy.py — the hundred-line ORM, doing the four things an ORM does.

Run it:  python3 examples/demo_toy.py

It generates DDL from class attributes, inserts objects, reads rows back as
objects, and proves the identity map both by object identity and by the number
of statements it did NOT send.
"""

from __future__ import annotations

import sqlite3

from tiny_orm import Column, Model, Session


class Member(Model):
    __table__ = "members"
    id = Column("INTEGER", primary_key=True)
    name = Column("TEXT")
    email = Column("TEXT")


class Book(Model):
    __table__ = "books"
    id = Column("INTEGER", primary_key=True)
    title = Column("TEXT")
    author = Column("TEXT")
    copies = Column("INTEGER")


def rule(label: str) -> None:
    print()
    print(label)
    print("-" * len(label))


def main() -> None:
    connection = sqlite3.connect(":memory:")
    session = Session(connection)

    rule("1. The class declaration IS the schema")
    print(Member.create_table_sql())
    print(Book.create_table_sql())
    session.create_all(Member, Book)

    rule("2. add() makes an object pending — no SQL yet")
    ada = Member(name="Ada Okonkwo", email="ada@library.test")
    bruno = Member(name="Bruno Sartori", email="bruno@library.test")
    session.add(ada)
    session.add(bruno)
    session.add(Book(title="The C Programming Language", author="Kernighan and Ritchie", copies=3))
    session.add(Book(title="Design Patterns", author="Gamma and others", copies=2))
    before = len(session.statements)
    print(f"objects pending: {len(session.pending)}")
    print(f"statements emitted so far: {before} (the two CREATE TABLEs above)")
    print(f"ada.id before flush: {ada.id}")

    rule("3. flush() turns pending objects into INSERTs")
    session.flush()
    for statement in session.statements[before:]:
        print(f"    {statement}")
    print(f"ada.id after flush: {ada.id}  <- the database decided this, not you")
    session.commit()

    rule("4. Rows map back into objects")
    everyone = session.select(Member)
    for member in everyone:
        print(f"    {member}")

    rule("5. The identity map: the same row is the same object")
    before = len(session.statements)
    first = session.get(Member, 1)
    second = session.get(Member, 1)
    print(f"first is second      : {first is second}")
    print(f"first is ada         : {first is ada}")
    print(f"statements emitted   : {len(session.statements) - before}")
    print("Both lookups were answered from the identity map, so no SELECT was sent.")

    rule("6. Why the identity map matters")
    first.name = "Ada O."
    print(f"changed via `first`, read via `second`: {second.name}")
    print("Without an identity map these would be two objects and one of the")
    print("two edits would be silently thrown away on the next write.")

    rule("7. Every statement this session sent")
    for number, statement in enumerate(session.statements, start=1):
        print(f"    {number:>2}. {statement}")

    connection.close()


if __name__ == "__main__":
    main()

"""demo_unit_of_work.py — object states, autoflush, flush versus commit.

Run it:  python3 examples/demo_unit_of_work.py

Almost every confusing ORM error is really a question about one of three
things: which state an object is in, when the flush happened, and whether the
session that loaded the object is still open. This script makes all three
visible, using a real file-backed database so that a genuinely separate
connection can be asked what it can see.

The database is created in a temporary directory and deleted on the way out.
"""

from __future__ import annotations

import re
import sqlite3
import tempfile
from pathlib import Path

from sqlalchemy import create_engine, inspect, select
from sqlalchemy.orm import Session, selectinload
from sqlalchemy.orm.exc import DetachedInstanceError

from counting import QueryCounter
from library import build_engine
from models import Member


def rule(label: str) -> None:
    print()
    print(label)
    print("-" * len(label))


def first_line(error: Exception) -> str:
    """The message, with the object's memory address replaced so output is stable."""
    return re.sub(r"0x[0-9a-f]+", "0xADDR", str(error).splitlines()[0])


def state_of(instance) -> str:
    state = inspect(instance)
    for name in ("transient", "pending", "persistent", "deleted", "detached"):
        if getattr(state, name):
            return name
    return "unknown"


def peek_with_a_second_connection(path: Path) -> list[str]:
    """Read the members table through a connection SQLAlchemy knows nothing about."""
    connection = sqlite3.connect(path)
    try:
        rows = connection.execute("SELECT name FROM members ORDER BY id").fetchall()
        return [row[0] for row in rows]
    finally:
        connection.close()


def main() -> None:
    workdir = Path(tempfile.mkdtemp(prefix="day093-"))
    path = workdir / "library.db"
    try:
        engine = build_engine(f"sqlite:///{path}")

        rule("1. The four states of a mapped object")
        session = Session(engine)
        member = Member(name="Grace Mensah", email="grace@library.test")
        print(f"just constructed          -> {state_of(member)}")
        session.add(member)
        print(f"after session.add()       -> {state_of(member)}")
        session.flush()
        print(f"after session.flush()     -> {state_of(member)}   id={member.id}")
        session.commit()
        print(f"after session.commit()    -> {state_of(member)}")
        session.close()
        print(f"after session.close()     -> {state_of(member)}")

        rule("2. flush is not commit — asked of a second, independent connection")
        session = Session(engine)
        session.add(Member(name="Hana Ito", email="hana@library.test"))
        print(f"before flush, other connection sees: {peek_with_a_second_connection(path)[-1]!r} last")
        with QueryCounter(engine) as counted:
            session.flush()
        print("flush emitted:")
        print(counted.report())
        seen = peek_with_a_second_connection(path)
        print(f"after flush, other connection sees : {len(seen)} members, last {seen[-1]!r}")
        print("  The INSERT was sent. The transaction is open. Nobody else can see it.")
        session.commit()
        seen = peek_with_a_second_connection(path)
        print(f"after commit, other connection sees: {len(seen)} members, last {seen[-1]!r}")
        session.close()

        rule("3. Autoflush — a query flushes your pending work first")
        session = Session(engine)
        session.add(Member(name="Ivan Petrov", email="ivan@library.test"))
        print("added one pending Member, then ran an unrelated SELECT:")
        with QueryCounter(engine) as counted:
            session.scalars(select(Member).where(Member.name.like("I%"))).all()
        print(counted.report())
        print("  The INSERT was emitted first, so the SELECT could see it.")
        print("  That is autoflush, and it is why SQL appears at lines you never wrote.")
        session.rollback()
        session.close()

        rule("4. DetachedInstanceError, provoked on a scalar attribute")
        session = Session(engine)
        ada = session.get(Member, 1)
        session.commit()
        session.close()
        try:
            print(ada.name)
        except DetachedInstanceError as error:
            print("raised DetachedInstanceError:")
            print(f"    {first_line(error)}")
        print("  commit() expired every attribute; close() removed the connection")
        print("  that would have refreshed them. Nothing is left to read.")

        print("  Fix A — tell the session not to expire on commit:")
        with Session(engine, expire_on_commit=False) as session:
            ada = session.get(Member, 1)
            session.commit()
        print(f"    ada.name after close: {ada.name!r}")

        rule("5. DetachedInstanceError, provoked on a relationship")
        with Session(engine) as session:
            ada = session.get(Member, 1)
        try:
            print(len(ada.loans))
        except DetachedInstanceError as error:
            print("raised DetachedInstanceError:")
            print(f"    {first_line(error)}")
        print("  A lazy relationship is a SELECT waiting to happen, and the session")
        print("  it was waiting for is gone.")

        print("  Fix B — load the relationship while the session is still open:")
        with Session(engine) as session:
            ada = session.scalars(
                select(Member).options(selectinload(Member.loans)).where(Member.id == 1)
            ).one()
        print(f"    len(ada.loans) after close: {len(ada.loans)}")
        print("  Fix A and Fix B answer different questions. A keeps loaded columns")
        print("  readable; B decides in advance which related rows you will need.")

        engine.dispose()
    finally:
        for leftover in sorted(workdir.glob("library.db*")):
            leftover.unlink()
        workdir.rmdir()
        print()
        print(f"temporary database removed: {not workdir.exists()}")


if __name__ == "__main__":
    main()

"""library.py — one engine, one schema, one fixed set of seed rows.

Every demo and every test in this lab builds its database from here, so every
number printed in `expected-output/` is reproducible: the same six members, the
same eight books, the same twenty-four loans, the same dates, every time.

The default URL is an in-memory SQLite database. Nothing is written to disk
unless you pass a path, which is why the lab leaves no database behind.
"""

from __future__ import annotations

from sqlalchemy import create_engine, insert
from sqlalchemy.engine import Engine

from models import Base, Book, Loan, Member, Tag, book_tags

MEMBERS = [
    (1, "Ada Okonkwo", "ada@library.test"),
    (2, "Bruno Sartori", "bruno@library.test"),
    (3, "Chen Wei", "chen@library.test"),
    (4, "Divya Ramanan", "divya@library.test"),
    (5, "Emeka Balogun", "emeka@library.test"),
    (6, "Farida Haddad", "farida@library.test"),
]

BOOKS = [
    (1, "978-0131103627", "The C Programming Language", "Kernighan and Ritchie", 3),
    (2, "978-0201633610", "Design Patterns", "Gamma and others", 2),
    (3, "978-0262033848", "Introduction to Algorithms", "Cormen and others", 4),
    (4, "978-1449355739", "Learning Python", "Mark Lutz", 2),
    (5, "978-0596007126", "Head First Design Patterns", "Freeman and Robson", 1),
    (6, "978-0132350884", "Clean Code", "Robert C. Martin", 3),
    (7, "978-0201616224", "The Pragmatic Programmer", "Hunt and Thomas", 2),
    (8, "978-0134685991", "Effective Java", "Joshua Bloch", 1),
]

TAGS = [(1, "classic"), (2, "python"), (3, "craft"), (4, "algorithms")]

BOOK_TAGS = [
    (1, 1),
    (2, 1),
    (2, 3),
    (3, 1),
    (3, 4),
    (4, 2),
    (5, 3),
    (6, 3),
    (7, 1),
    (7, 3),
    (8, 3),
]

# (id, book_id, member_id, borrowed_on, due_on, returned)
LOANS = [
    (1, 1, 1, "2026-05-04", "2026-05-25", True),
    (2, 2, 1, "2026-06-01", "2026-06-22", False),
    (3, 3, 1, "2026-06-08", "2026-06-29", False),
    (4, 1, 2, "2026-05-11", "2026-06-01", True),
    (5, 4, 2, "2026-06-15", "2026-07-06", False),
    (6, 5, 3, "2026-04-20", "2026-05-11", True),
    (7, 6, 3, "2026-05-18", "2026-06-08", True),
    (8, 7, 3, "2026-06-22", "2026-07-13", False),
    (9, 8, 3, "2026-07-01", "2026-07-22", False),
    (10, 2, 4, "2026-03-09", "2026-03-30", True),
    (11, 3, 4, "2026-04-13", "2026-05-04", True),
    (12, 6, 4, "2026-07-06", "2026-07-27", False),
    (13, 1, 5, "2026-02-16", "2026-03-09", True),
    (14, 4, 5, "2026-05-25", "2026-06-15", True),
    (15, 5, 5, "2026-06-29", "2026-07-20", False),
    (16, 7, 5, "2026-07-13", "2026-08-03", False),
    (17, 3, 6, "2026-01-19", "2026-02-09", True),
    (18, 8, 6, "2026-03-23", "2026-04-13", True),
    (19, 6, 6, "2026-05-04", "2026-05-25", True),
    (20, 2, 6, "2026-06-08", "2026-06-29", False),
    (21, 4, 6, "2026-07-20", "2026-08-10", False),
    (22, 5, 1, "2026-07-27", "2026-08-17", False),
    (23, 8, 2, "2026-08-03", "2026-08-24", False),
    (24, 7, 4, "2026-08-10", "2026-08-31", False),
]


def build_engine(url: str = "sqlite://", echo: bool = False) -> Engine:
    """Create the engine, create the schema, and load the fixed seed rows.

    The seed is loaded through Core `insert()` rather than the ORM on purpose:
    it keeps the setup out of every query count the demos take afterwards.
    """
    engine = create_engine(url, echo=echo)
    Base.metadata.create_all(engine)
    with engine.begin() as connection:
        connection.execute(
            insert(Member),
            [
                {"id": i, "name": n, "email": e}
                for i, n, e in MEMBERS
            ],
        )
        connection.execute(
            insert(Book),
            [
                {"id": i, "isbn": s, "title": t, "author": a, "copies": c}
                for i, s, t, a, c in BOOKS
            ],
        )
        connection.execute(
            insert(Tag), [{"id": i, "name": n} for i, n in TAGS]
        )
        connection.execute(
            insert(book_tags),
            [{"book_id": b, "tag_id": t} for b, t in BOOK_TAGS],
        )
        connection.execute(
            insert(Loan),
            [
                {
                    "id": i,
                    "book_id": b,
                    "member_id": m,
                    "borrowed_on": bo,
                    "due_on": d,
                    "returned": r,
                }
                for i, b, m, bo, d, r in LOANS
            ],
        )
    return engine

"""models.py — the same library domain, mapped with SQLAlchemy 2.0.

This is the modern declarative style: a DeclarativeBase subclass, `Mapped[...]`
annotations, and `mapped_column()`. The 1.x `declarative_base()` factory and the
`Query` object are legacy; they still work, and you will meet them in old code,
but nothing here uses them.

Compare each class with the CREATE TABLE you wrote by hand in Week 13. The
columns are the same columns and the constraints are the same constraints. What
is new is that the class is also a Python object with behaviour, and that a
relationship attribute stands where a join used to be.
"""

from __future__ import annotations

from sqlalchemy import CheckConstraint, Column, ForeignKey, Table, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    """One base per application. It owns the MetaData all tables register in."""


# A many-to-many needs a table with no class of its own: it carries nothing but
# the two foreign keys. This is the "secondary" table, and SQLAlchemy wants it
# as a Core Table rather than a mapped class precisely because it is not an
# entity — it has no identity worth talking about.
book_tags = Table(
    "book_tags",
    Base.metadata,
    Column("book_id", ForeignKey("books.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
)


class Member(Base):
    __tablename__ = "members"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(Text)
    email: Mapped[str] = mapped_column(Text, unique=True)

    loans: Mapped[list[Loan]] = relationship(
        back_populates="member", cascade="all, delete-orphan"
    )

    __table_args__ = (CheckConstraint("length(trim(name)) > 0", name="ck_member_name"),)

    def __repr__(self) -> str:
        return f"Member(id={self.id!r}, name={self.name!r})"


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(Text, unique=True)

    books: Mapped[list[Book]] = relationship(secondary=book_tags, back_populates="tags")

    def __repr__(self) -> str:
        return f"Tag(id={self.id!r}, name={self.name!r})"


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    isbn: Mapped[str] = mapped_column(Text, unique=True)
    title: Mapped[str] = mapped_column(Text)
    author: Mapped[str] = mapped_column(Text)
    copies: Mapped[int] = mapped_column(default=1)

    loans: Mapped[list[Loan]] = relationship(back_populates="book")
    tags: Mapped[list[Tag]] = relationship(secondary=book_tags, back_populates="books")

    __table_args__ = (CheckConstraint("copies >= 0", name="ck_book_copies"),)

    def __repr__(self) -> str:
        return f"Book(id={self.id!r}, title={self.title!r})"


class Loan(Base):
    __tablename__ = "loans"

    id: Mapped[int] = mapped_column(primary_key=True)
    book_id: Mapped[int] = mapped_column(ForeignKey("books.id"))
    member_id: Mapped[int] = mapped_column(ForeignKey("members.id"))
    borrowed_on: Mapped[str] = mapped_column(Text)
    due_on: Mapped[str] = mapped_column(Text)
    returned: Mapped[bool] = mapped_column(default=False)

    book: Mapped[Book] = relationship(back_populates="loans")
    member: Mapped[Member] = relationship(back_populates="loans")

    __table_args__ = (
        CheckConstraint("due_on >= borrowed_on", name="ck_loan_dates"),
    )

    def __repr__(self) -> str:
        return f"Loan(id={self.id!r}, book_id={self.book_id!r}, returned={self.returned!r})"

"""tiny_orm.py — a working object-relational mapper in under a hundred lines.

Build this before you touch SQLAlchemy. Everything the real library does is
here in miniature: columns declared as class attributes, DDL and DML generated
from that declaration, rows mapped back into objects, and an identity map so
that the same row fetched twice yields the *same* Python object.

Nothing here is clever. That is the point: once you have written the toy, the
real library stops being magic and becomes a much more careful version of code
you already understand.

Every statement this module emits is recorded on the session, so the tests can
assert on what was sent to the database rather than on what you hoped was sent.
"""

from __future__ import annotations

import sqlite3


class Column:
    """One mapped column. Declared as a class attribute on a Model subclass."""

    def __init__(self, sql_type: str, primary_key: bool = False) -> None:
        self.sql_type = sql_type
        self.primary_key = primary_key
        self.name: str | None = None

    def __set_name__(self, owner: type, name: str) -> None:
        # Python calls this at class-creation time and hands us the attribute
        # name, so a column never has to repeat its own name.
        self.name = name


class ModelMeta(type):
    """Collects the Column attributes of each subclass into __columns__."""

    def __new__(mcls, name, bases, namespace):
        cls = super().__new__(mcls, name, bases, namespace)
        cls.__columns__ = {
            key: value for key, value in namespace.items() if isinstance(value, Column)
        }
        return cls


class Model(metaclass=ModelMeta):
    """Base class for mapped objects. Subclasses set __table__ and Columns."""

    __table__: str = ""

    def __init__(self, **values) -> None:
        unknown = set(values) - set(type(self).__columns__)
        if unknown:
            raise TypeError(f"{type(self).__name__} has no column(s): {sorted(unknown)}")
        for column_name in type(self).__columns__:
            # A plain instance attribute shadows the Column class attribute,
            # which is the whole trick: after __init__, obj.title is the value.
            setattr(self, column_name, values.get(column_name))

    @classmethod
    def primary_key_name(cls) -> str:
        for column_name, column in cls.__columns__.items():
            if column.primary_key:
                return column_name
        raise TypeError(f"{cls.__name__} declares no primary key")

    @classmethod
    def create_table_sql(cls) -> str:
        pieces = []
        for column_name, column in cls.__columns__.items():
            piece = f"{column_name} {column.sql_type}"
            if column.primary_key:
                piece += " PRIMARY KEY"
            pieces.append(piece)
        return f"CREATE TABLE {cls.__table__} ({', '.join(pieces)})"

    def __repr__(self) -> str:
        shown = ", ".join(
            f"{name}={getattr(self, name)!r}" for name in type(self).__columns__
        )
        return f"{type(self).__name__}({shown})"


class Session:
    """A unit of work: pending objects, an identity map, and a flush."""

    def __init__(self, connection: sqlite3.Connection) -> None:
        self.connection = connection
        self.identity_map: dict[tuple[type, object], Model] = {}
        self.pending: list[Model] = []
        self.statements: list[str] = []

    def execute(self, sql: str, parameters: tuple = ()) -> sqlite3.Cursor:
        self.statements.append(sql)
        return self.connection.execute(sql, parameters)

    def create_all(self, *model_classes: type[Model]) -> None:
        for model_class in model_classes:
            self.execute(model_class.create_table_sql())

    def add(self, instance: Model) -> None:
        """Make the object pending. No SQL is emitted here — that is the point."""
        self.pending.append(instance)

    def flush(self) -> None:
        """Turn every pending object into an INSERT. Still no commit."""
        for instance in self.pending:
            model_class = type(instance)
            column_names = list(model_class.__columns__)
            placeholders = ", ".join("?" for _ in column_names)
            sql = (
                f"INSERT INTO {model_class.__table__} "
                f"({', '.join(column_names)}) VALUES ({placeholders})"
            )
            values = tuple(getattr(instance, name) for name in column_names)
            cursor = self.execute(sql, values)
            key_name = model_class.primary_key_name()
            if getattr(instance, key_name) is None:
                setattr(instance, key_name, cursor.lastrowid)
            self.identity_map[(model_class, getattr(instance, key_name))] = instance
        self.pending.clear()

    def commit(self) -> None:
        self.flush()
        self.connection.commit()
        self.statements.append("COMMIT")

    def _instance_from_row(self, model_class: type[Model], row: tuple) -> Model:
        column_names = list(model_class.__columns__)
        values = dict(zip(column_names, row))
        key = (model_class, values[model_class.primary_key_name()])
        if key in self.identity_map:
            return self.identity_map[key]
        instance = model_class(**values)
        self.identity_map[key] = instance
        return instance

    def get(self, model_class: type[Model], key_value) -> Model | None:
        """Fetch by primary key. A hit in the identity map emits NO SQL."""
        key = (model_class, key_value)
        if key in self.identity_map:
            return self.identity_map[key]
        column_names = ", ".join(model_class.__columns__)
        sql = (
            f"SELECT {column_names} FROM {model_class.__table__} "
            f"WHERE {model_class.primary_key_name()} = ?"
        )
        row = self.execute(sql, (key_value,)).fetchone()
        if row is None:
            return None
        return self._instance_from_row(model_class, row)

    def select(self, model_class: type[Model], **equals) -> list[Model]:
        column_names = ", ".join(model_class.__columns__)
        sql = f"SELECT {column_names} FROM {model_class.__table__}"
        if equals:
            sql += " WHERE " + " AND ".join(f"{name} = ?" for name in equals)
        rows = self.execute(sql, tuple(equals.values())).fetchall()
        return [self._instance_from_row(model_class, row) for row in rows]

lesson_id: D093
day: 93
kind: guided-build
languages: [python, sql, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-093-orms-and-sqlalchemy
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/python3 -c "import sqlalchemy; print(sqlalchemy.__version__)"
run_commands:
  - export PYTHONPATH=examples
  - .venv/bin/python3 examples/demo_toy.py
  - .venv/bin/python3 examples/demo_sqlalchemy.py
  - .venv/bin/python3 examples/demo_unit_of_work.py
  - .venv/bin/python3 examples/demo_n_plus_one.py
  - .venv/bin/python3 examples/demo_bulk.py
  - .venv/bin/pytest starter -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - "find . -type d -name '__pycache__' -not -path './.venv/*' -prune -exec rm -rf -- {} +"
  - rm -rf starter/.pytest_cache
  - 'rm -rf .venv  # optional: removes the lab virtual environment'
  - 'git checkout -- starter/  # optional: reset your exercise work'
requires_network: true
requires_api_key: false
estimated_minutes: 35
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, SQLAlchemy 2.0.51, pytest 9.1.1, SQLite library 3.53.3 via Python, bash 3.2.57 — bash tests/run_tests.sh -> 87 checks, 0 failure(s), exit 0; pytest starter -q -> 1 passed, 9 skipped; all five demos exit 0 and match expected-output/ byte for byte. Harness proved to fail: removing the selectinload from examples/demo_n_plus_one.py reports 87 checks, 2 failure(s) with a non-zero exit, caught once by the statement count and once by the capture comparison. Network is needed once to install the two pinned packages; nothing else in the lab opens a socket, and section 9 trips the guard in starter/conftest.py deliberately to prove it is armed. Alembic is deliberately NOT installed and section 1 asserts it is still absent, because the lesson states plainly that no Alembic output is reproduced.'

# Dependencies for the Day 093 lab

Two packages, both free and open source, both installed from the Python
Package Index with `pip`, both running entirely on your own machine.

| Package | Pinned version | Why this lab needs it |
| --- | --- | --- |
| `SQLAlchemy` | `2.0.51` | The library the lesson is about. It provides both layers the lesson separates: Core, an SQL expression language, and the ORM built on top of it. |
| `pytest` | `9.1.1` | The test runner from Days 071–074. Nothing new today except what it is pointed at — statement counts rather than return values. |

One package arrives that nobody asked for. `typing_extensions` (4.16.0 here)
is a SQLAlchemy dependency and is deliberately **not** pinned: SQLAlchemy
itself constrains which versions it accepts, and pinning a transitive
dependency separately is how you eventually get an unsolvable conflict.

The pinned numbers were read from the installed packages rather than assumed:

```bash
.venv/bin/python3 -c "from importlib.metadata import version; print(version('SQLAlchemy'), version('pytest'))"
```

On the authoring machine, on 16 August 2026, that printed `2.0.51 9.1.1`, and
section 1 of `tests/run_tests.sh` reprints the installed version and compares
it against `requirements.txt` — so a mismatch is reported at the top of the
run rather than discovered later as a mysterious count.

## What is deliberately absent

**Alembic**, SQLAlchemy's migration tool, is **not installed and not used
here.** The lesson describes what it does from its documentation and says
plainly that no output is reproduced for it. That is not an oversight — it is
a scope decision, because migrations are a Day 88 topic that deserves its own
treatment against a real ORM rather than a paragraph at the end of this one.

Section 1 of the test suite checks that Alembic is still absent. If somebody
installs it into this environment, the suite fails, because at that point the
lesson's honesty note would need rewriting rather than quietly becoming false.

**No database driver is installed either**, and none is needed. Every database
in this lab is SQLite, reached through `sqlite3` in the Python standard
library, which SQLAlchemy uses as its default driver for the `sqlite://` URL.
Nothing here talks to PostgreSQL or MySQL. Where the lesson mentions them, it
mentions them as documented behaviour and claims no measurement.

## Licences

SQLAlchemy is distributed under the MIT licence and pytest under the MIT
licence, each stated on the project's own documentation. Both are maintained
in the open, cost nothing, and need no account, no key and no signup —
personally or commercially.

## One-time install

From the lab directory:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
.venv/bin/python3 -c "import sqlalchemy; print(sqlalchemy.__version__)"
```

Expect `2.0.51`. Day 43 covered `python3 -m venv` in full; this is the same
pattern. The environment lives in `.venv/` inside the lab, is already excluded
from version control, and can be deleted at any time with `rm -rf .venv`.

## Network

Installing needs the network, once. **Nothing else in this lab does.** Every
database here is either in memory (`sqlite://`) or a file in a temporary
directory that is deleted on the way out, and `starter/conftest.py` arms a
guard that raises if anything tries to open a socket. Section 9 of
`tests/run_tests.sh` proves that guard is armed by tripping it deliberately.

## Running without a lab-local environment

If you already have SQLAlchemy 2.x and pytest available in an environment you
have activated, the test runner will find them on your `PATH`. You can also
point it at specific binaries:

```bash
PYTHON=/path/to/python3 PYTEST=/path/to/pytest bash tests/run_tests.sh
```

The runner checks that SQLAlchemy is importable from the interpreter it
resolved, and **stops with install instructions rather than skipping checks
quietly** if it is not. A test suite that silently skips the only thing it was
written to test is worse than one that fails.

## If your SQLAlchemy is a different version

Every statement count in `expected-output/` was captured on 2.0.51. The counts
in sections 2 to 5 of the harness are properties of how the ORM works and
should hold across 2.x. The **bulk** counts in section 6 are the ones most
likely to move, because how many rows SQLAlchemy packs into one `executemany`
is an implementation decision that has changed across releases and varies by
dialect. `expected-output/FIELDS.md` says exactly which numbers are structural
and which are version-specific.

SQLAlchemy==2.0.51
pytest==9.1.1

"""Test configuration for the starter exercises.

Two jobs, both small.

1. Put the lab's `examples/` directory on `sys.path`, so `queries.py` and the
   tests can import `models`, `library` and `counting` without any packaging
   ceremony. Those three modules are shared infrastructure — the domain, the
   fixed seed data and the statement counter — and copying them into
   `starter/` would mean two versions to keep in step.

2. Arm a guard that turns any attempt to open a network socket into a loud
   failure. Installing SQLAlchemy needs the network once. Nothing in this lab
   does, and a test that quietly reaches the internet is a test that fails in
   a tunnel for reasons nobody can reproduce.
"""

from __future__ import annotations

import socket
import sys
from pathlib import Path

LAB_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(LAB_DIR / "examples"))
sys.path.insert(0, str(Path(__file__).resolve().parent))


class NetworkAccessAttempted(RuntimeError):
    """Raised if anything in this lab tries to open a connection."""


def _refuse(*args, **kwargs):
    raise NetworkAccessAttempted(
        "This lab runs entirely offline against in-memory and temporary SQLite "
        "databases. Something tried to open a network connection."
    )


socket.socket.connect = _refuse
socket.create_connection = _refuse

[pytest]
testpaths = .
python_files = test_*.py
# Any warning SQLAlchemy raises here is a real signal about how the ORM was
# used, so nothing is filtered. If you see one, read it — the library's
# warnings about relationship conflicts and implicit coercions are some of the
# most useful diagnostics it has.
filterwarnings =
    error::sqlalchemy.exc.SAWarning

"""Your ORM work — a working skeleton with nine exercises.

This file RUNS right now, and every function in it returns the right answer.
That is the point. **None of the exercises below are about correctness.** They
are about what the ORM sent to the database to get there, which you cannot
see by reading the Python and which the tests measure for you.

Prove the baseline before you change anything. From the lab directory:

    python3 -m venv .venv
    .venv/bin/pip install -r requirements/requirements.txt
    .venv/bin/pytest starter -q

You should see one test pass and nine skipped. Each skipped test names the
exercise that makes it pass. Work through them in order; after each one,
rerun the command above and delete the `@pytest.mark.skip` line from the test
you just satisfied.

What is here already is deliberately the version somebody writes on their
first afternoon with an ORM:

  * filtering and joining in Python, because the objects are right there;
  * touching a relationship inside a loop, which is the N+1 problem written
    in a way that looks completely innocent;
  * reading attributes after the session closed, which raises;
  * changing rows one object at a time when a single statement would do.

Every one of those runs. Every one of those returns the correct answer. The
tests fail on the STATEMENT COUNT, not on the values — which is the habit this
whole lab exists to build.

The reference implementations live in `examples/`. Run them whenever you want
to see where you are heading:

    .venv/bin/python3 examples/demo_n_plus_one.py
    .venv/bin/python3 examples/demo_unit_of_work.py
"""

from __future__ import annotations

import sqlite3
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

from counting import QueryCounter
from library import build_engine
from models import Book, Loan, Member


# ---------------------------------------------------------------------------
# EXERCISE 1 — filter in the database, not in Python
# ---------------------------------------------------------------------------
def books_with_at_least(session: Session, copies: int) -> list[str]:
    """Titles of every book with at least `copies` copies, alphabetically.

    Right answer, wrong query. This pulls the whole books table across the
    wire and then throws most of it away in Python. On eight books nobody
    notices. On eight million it is the outage.

    EXERCISE 1: rewrite the body as ONE `select()` that does the filtering
                and the ordering in SQL. The shape you want is

                    select(Book.title).where(Book.copies >= copies)
                                      .order_by(Book.title)

                and then `session.scalars(...).all()` to run it.

                The test asserts that exactly one statement is emitted and
                that the emitted SQL contains both WHERE and ORDER BY.
    """
    everything = session.scalars(select(Book)).all()
    matching = [book.title for book in everything if book.copies >= copies]
    return sorted(matching)


# ---------------------------------------------------------------------------
# EXERCISE 2 — let the database do the join and the counting
# ---------------------------------------------------------------------------
def open_loan_counts(session: Session) -> list[tuple[str, int]]:
    """(member name, number of unreturned loans), busiest first, then by name.

    EXERCISE 2: replace the two queries and the Python bookkeeping with ONE
                `select()` that joins members to loans, filters on
                `Loan.returned.is_(False)`, groups by `Member.id`, and orders
                by the count descending then the name. You will want
                `func.count` — import it with `from sqlalchemy import func`.

                Careful: a plain JOIN drops members with no open loans. This
                naive version keeps them at zero, and the test checks that all
                six members appear. Use `outerjoin` and count `Loan.id`, which
                counts non-NULL values and therefore gives 0 rather than 1 for
                a member with no matching loan row.

                The test asserts exactly one statement and all six members.
    """
    members = session.scalars(select(Member).order_by(Member.id)).all()
    loans = session.scalars(select(Loan).where(Loan.returned.is_(False))).all()
    tally: dict[int, int] = {member.id: 0 for member in members}
    for loan in loans:
        tally[loan.member_id] += 1
    named = [(member.name, tally[member.id]) for member in members]
    return sorted(named, key=lambda pair: (-pair[1], pair[0]))


# ---------------------------------------------------------------------------
# EXERCISE 3 — the N+1, fixed with selectinload
# ---------------------------------------------------------------------------
def member_loan_totals(session: Session) -> list[tuple[str, int]]:
    """(member name, total loans ever), in id order.

    This is the N+1 problem, and notice how ordinary it looks. `member.loans`
    reads like a list attribute. It is a SELECT, issued the first time you
    touch it, once per member.

    EXERCISE 3: keep the loop — the loop is fine and readable — and change
                only the query that feeds it, so the relationship is loaded up
                front:

                    from sqlalchemy.orm import selectinload
                    select(Member).options(selectinload(Member.loans))
                                  .order_by(Member.id)

                The test asserts EXACTLY 2 statements: one for the members,
                one for all their loans. Not 7. And not 1 — selectinload
                deliberately uses a second query rather than a join.
    """
    members = session.scalars(select(Member).order_by(Member.id)).all()
    return [(member.name, len(member.loans)) for member in members]


# ---------------------------------------------------------------------------
# EXERCISE 4 — the many-to-one N+1, fixed with joinedload
# ---------------------------------------------------------------------------
def loan_titles(session: Session) -> list[str]:
    """The book title for every loan, in loan id order.

    Same defect, different relationship. `loan.book` is many-to-one: every
    loan has exactly one book, so a JOIN cannot multiply the rows, which makes
    this the case joinedload was designed for.

    EXERCISE 4: load the relationship with `joinedload(Loan.book)`:

                    from sqlalchemy.orm import joinedload
                    select(Loan).options(joinedload(Loan.book)).order_by(Loan.id)

                The test asserts EXACTLY 1 statement. Note you do NOT need
                `.unique()` here, because a many-to-one join returns one row
                per loan. Exercise 3's collection would have needed it.
    """
    loans = session.scalars(select(Loan).order_by(Loan.id)).all()
    return [loan.book.title for loan in loans]


# ---------------------------------------------------------------------------
# EXERCISE 5 — flush is not commit
# ---------------------------------------------------------------------------
def flush_then_commit(path: Path) -> tuple[int, int, int]:
    """Add one member; report what a SEPARATE connection can see at each step.

    Returns (before_flush, after_flush, after_commit) — three row counts, each
    read through `peek(path)`, which opens its own sqlite3 connection that
    SQLAlchemy knows nothing about.

    This version calls `commit()` straight away, so it never shows the
    interesting middle state.

    EXERCISE 5: split the write into an explicit `session.flush()` and then a
                `session.commit()`, taking a `peek(path)` reading between
                them. The result should be (6, 6, 7): the INSERT really was
                sent at the flush, and the other connection really could not
                see it until the commit, because it was inside an open
                transaction.

                The test asserts that triple exactly.
    """
    engine = build_engine(f"sqlite:///{path}")
    try:
        before = peek(path)
        with Session(engine) as session:
            session.add(Member(name="Grace Mensah", email="grace@library.test"))
            session.commit()
            after_flush = peek(path)
        after_commit = peek(path)
        return (before, after_flush, after_commit)
    finally:
        engine.dispose()


def peek(path: Path) -> int:
    """How many members a connection outside SQLAlchemy's control can see."""
    connection = sqlite3.connect(path)
    try:
        return connection.execute("SELECT count(*) FROM members").fetchone()[0]
    finally:
        connection.close()


# ---------------------------------------------------------------------------
# EXERCISE 6 — the four object states
# ---------------------------------------------------------------------------
def state_sequence(engine: Engine) -> list[str]:
    """The state of one Member after construct, add, flush and close.

    Nearly every confusing ORM error is a question about which of these an
    object is in, so being able to name them on demand is worth the five
    minutes.

    EXERCISE 6: return the four state names as strings, in order, by asking
                SQLAlchemy rather than by hard-coding them:

                    from sqlalchemy import inspect
                    inspect(member).transient / .pending / .persistent / .detached

                Write a small helper that returns the name of whichever flag
                is True. The expected answer is
                ["transient", "pending", "persistent", "detached"] — but the
                test also checks you did not simply return that literal list,
                by running the same helper against an object it manipulates
                itself.
    """
    session = Session(engine)
    member = Member(name="Hana Ito", email="hana@library.test")
    session.add(member)
    session.flush()
    session.commit()
    session.close()
    return ["unknown", "unknown", "unknown", "unknown"]


def state_of(instance: object) -> str:
    """Name the state of one mapped object.

    EXERCISE 6 (part two): implement this. `state_sequence` should call it
    four times, and the test calls it directly on objects of its own.
    """
    return "unknown"


# ---------------------------------------------------------------------------
# EXERCISE 7 — DetachedInstanceError, fixed by not expiring
# ---------------------------------------------------------------------------
def name_after_close(engine: Engine) -> str:
    """Read a member's name AFTER the session that loaded it has closed.

    Run this as it stands and it raises DetachedInstanceError. That is not a
    bug being demonstrated for fun — it is the single most common wall a
    beginner hits, and it has a precise cause: `commit()` expires every loaded
    attribute so the next read will be fresh, and `close()` then takes away
    the connection that read would have used.

    EXERCISE 7: fix it WITHOUT loading anything extra, by telling the session
                not to expire attributes on commit:

                    with Session(engine, expire_on_commit=False) as session:

                Then `member.name` is still readable afterwards, because the
                value that was already loaded was never thrown away.

                The test asserts the name comes back as 'Ada Okonkwo' and no
                exception escapes.
    """
    with Session(engine) as session:
        member = session.get(Member, 1)
        session.commit()
    return member.name


# ---------------------------------------------------------------------------
# EXERCISE 8 — DetachedInstanceError, fixed by eager loading
# ---------------------------------------------------------------------------
def loan_count_after_close(engine: Engine) -> int:
    """How many loans member 1 has, counted AFTER the session closed.

    This raises too, and for a related but genuinely different reason: a lazy
    relationship is a SELECT waiting to happen, and the session it was waiting
    for is gone. `expire_on_commit=False` would NOT save you here, because the
    relationship was never loaded in the first place — there is nothing to
    keep.

    EXERCISE 8: fix it by deciding in advance that you need the loans, and
                loading them while the session is open:

                    select(Member).options(selectinload(Member.loans))
                                  .where(Member.id == 1)

                then `session.scalars(...).one()`.

                The test asserts the answer is 4, and — the part that matters —
                that ZERO statements are emitted after the session closes,
                because there is no session left to emit them.
    """
    with Session(engine) as session:
        member = session.get(Member, 1)
    return len(member.loans)


# ---------------------------------------------------------------------------
# EXERCISE 9 — when to stop using the ORM
# ---------------------------------------------------------------------------
def close_all_open_loans(session: Session) -> int:
    """Mark every unreturned loan as returned. Return the number changed.

    The ORM way below is correct and it is readable, and it does something you
    may not want: it SELECTs every matching row and builds a Python object for
    each one, purely so it can set a flag. With thirteen rows that is nothing.
    With a million it is a memory incident.

    EXERCISE 9: replace the body with one Core UPDATE:

                    from sqlalchemy import update
                    result = session.execute(
                        update(Loan).where(Loan.returned.is_(False))
                                    .values(returned=True)
                    )
                    return result.rowcount

                The test asserts exactly ONE cursor execution and a rowcount
                of 13. Read `examples/demo_bulk.py` afterwards for the price
                you just paid: nothing you wrote in Python runs for those
                rows, because no object was ever created.
    """
    changed = 0
    for loan in session.scalars(select(Loan).where(Loan.returned.is_(False))):
        loan.returned = True
        changed += 1
    session.flush()
    return changed


__all__ = [
    "QueryCounter",
    "books_with_at_least",
    "build_engine",
    "close_all_open_loans",
    "flush_then_commit",
    "loan_count_after_close",
    "loan_titles",
    "member_loan_totals",
    "name_after_close",
    "open_loan_counts",
    "peek",
    "state_of",
    "state_sequence",
]

"""Your exercise suite. One test passes now; nine are waiting for you.

Run it from the lab directory:

    .venv/bin/pytest starter -q

Each skipped test names the exercise in `queries.py` that makes it pass. Do
the exercise, delete that test's `@pytest.mark.skip(...)` line, rerun. When
all nine are green, you have written the ORM code the lesson argues for and
you have proved it by counting statements rather than by trusting the output.

**Read this before you start, because it is the transferable lesson:** almost
every assertion below is on a COUNT of statements, never on a duration. A
timing assertion is a flake waiting for a loaded machine, and it names no
cause — "this took 240 ms" is a mood. A count is deterministic, identical on
every machine, and it names the defect directly: "this loop issued seven
queries where two would do" is a bug report you can act on.
"""

from __future__ import annotations

import tempfile
from pathlib import Path

import pytest
from counting import QueryCounter
from library import build_engine
from queries import (
    books_with_at_least,
    close_all_open_loans,
    flush_then_commit,
    loan_count_after_close,
    loan_titles,
    member_loan_totals,
    name_after_close,
    open_loan_counts,
    state_of,
    state_sequence,
)
from sqlalchemy.orm import Session


@pytest.fixture
def engine():
    made = build_engine()
    yield made
    made.dispose()


@pytest.fixture
def session(engine):
    with Session(engine) as opened:
        yield opened


def test_the_seed_is_what_we_think_it_is(session) -> None:
    """This one passes already. It is your green baseline: if it ever fails,
    the problem is your setup rather than your code."""
    from models import Book, Loan, Member, Tag
    from sqlalchemy import func, select

    counts = {
        "members": session.scalar(select(func.count()).select_from(Member)),
        "books": session.scalar(select(func.count()).select_from(Book)),
        "tags": session.scalar(select(func.count()).select_from(Tag)),
        "loans": session.scalar(select(func.count()).select_from(Loan)),
    }
    assert counts == {"members": 6, "books": 8, "tags": 4, "loans": 24}


@pytest.mark.skip(reason="Exercise 1: filter in the database, not in Python")
def test_books_with_at_least_filters_in_sql(engine, session) -> None:
    with QueryCounter(engine) as counted:
        titles = books_with_at_least(session, 3)

    assert titles == [
        "Clean Code",
        "Introduction to Algorithms",
        "The C Programming Language",
    ]
    assert len(counted) == 1, f"expected one statement, got {len(counted)}"
    emitted = counted.statements[0].upper()
    assert "WHERE" in emitted, "the filter is still happening in Python"
    assert "ORDER BY" in emitted, "the sort is still happening in Python"


@pytest.mark.skip(reason="Exercise 2: one grouped outer join instead of two queries")
def test_open_loan_counts_is_one_grouped_query(engine, session) -> None:
    with QueryCounter(engine) as counted:
        rows = open_loan_counts(session)

    assert len(counted) == 1, f"expected one statement, got {len(counted)}"
    assert len(rows) == 6, "every member must appear, including any with none open"
    assert rows[0] == ("Ada Okonkwo", 3)
    assert sum(count for _, count in rows) == 13
    emitted = counted.statements[0].upper()
    assert "GROUP BY" in emitted, "the grouping is still happening in Python"
    assert "JOIN" in emitted, "the join is still happening in Python"


@pytest.mark.skip(reason="Exercise 3: fix the N+1 with selectinload")
def test_member_loan_totals_is_exactly_two_statements(engine, session) -> None:
    with QueryCounter(engine) as counted:
        rows = member_loan_totals(session)

    assert rows == [
        ("Ada Okonkwo", 4),
        ("Bruno Sartori", 3),
        ("Chen Wei", 4),
        ("Divya Ramanan", 4),
        ("Emeka Balogun", 4),
        ("Farida Haddad", 5),
    ]
    assert len(counted) == 2, (
        f"expected exactly 2 statements, got {len(counted)}. "
        "7 means the relationship is still lazy; 1 means you used joinedload, "
        "which works but is not what selectinload does."
    )


@pytest.mark.skip(reason="Exercise 4: fix the many-to-one N+1 with joinedload")
def test_loan_titles_is_exactly_one_statement(engine, session) -> None:
    with QueryCounter(engine) as counted:
        titles = loan_titles(session)

    assert len(titles) == 24
    assert titles[0] == "The C Programming Language"
    assert len(set(titles)) == 8
    assert len(counted) == 1, (
        f"expected exactly 1 statement, got {len(counted)}. A many-to-one "
        "joinedload cannot multiply rows, so one query is the whole job."
    )


@pytest.mark.skip(reason="Exercise 5: separate the flush from the commit")
def test_flush_is_not_commit() -> None:
    workdir = Path(tempfile.mkdtemp(prefix="day093-starter-"))
    try:
        result = flush_then_commit(workdir / "library.db")
    finally:
        for leftover in sorted(workdir.glob("library.db*")):
            leftover.unlink()
        workdir.rmdir()

    assert result == (6, 6, 7), (
        f"expected (6, 6, 7), got {result}. The middle number is the point: "
        "the INSERT had been sent, and an outside connection still could not "
        "see it, because the transaction was open."
    )


@pytest.mark.skip(reason="Exercise 6: name the four object states")
def test_state_sequence_and_state_of(engine) -> None:
    assert state_sequence(engine) == [
        "transient",
        "pending",
        "persistent",
        "detached",
    ]

    # And prove `state_of` really inspects, rather than returning a script.
    from models import Member

    fresh = Member(name="Ivan Petrov", email="ivan@library.test")
    assert state_of(fresh) == "transient"
    with Session(engine) as opened:
        opened.add(fresh)
        assert state_of(fresh) == "pending"
        opened.flush()
        assert state_of(fresh) == "persistent"
        opened.rollback()


@pytest.mark.skip(reason="Exercise 7: fix the detached read with expire_on_commit")
def test_name_readable_after_close(engine) -> None:
    assert name_after_close(engine) == "Ada Okonkwo"


@pytest.mark.skip(reason="Exercise 8: fix the detached relationship with selectinload")
def test_loans_readable_after_close(engine) -> None:
    with QueryCounter(engine) as counted:
        total = loan_count_after_close(engine)
        emitted_inside = len(counted)

    assert total == 4
    # Everything must have been loaded before the session closed. If the
    # relationship were still lazy this would have raised rather than counted.
    assert emitted_inside == 2, (
        f"expected 2 statements (members, then their loans), got {emitted_inside}"
    )


@pytest.mark.skip(reason="Exercise 9: one Core UPDATE instead of an object loop")
def test_close_all_open_loans_is_one_statement(engine, session) -> None:
    with QueryCounter(engine) as counted:
        changed = close_all_open_loans(session)
    session.rollback()

    assert changed == 13
    assert len(counted) == 1, (
        f"expected exactly 1 cursor execution, got {len(counted)}. Two means "
        "you are still SELECTing the rows in order to change them."
    )
    assert counted.statements[0].upper().startswith("UPDATE")

#!/usr/bin/env bash
# Tests for the Day 093 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# This harness proves the claims the lesson makes, and it proves nearly all of
# them by COUNTING THE STATEMENTS the ORM emitted rather than by checking that
# the answer looked right. That distinction is the lab:
#
#   * the toy ORM generates its own DDL and DML, maps rows back into objects,
#     and answers a repeat lookup from its identity map with NO SQL at all;
#   * SQLAlchemy 2.0's declarative models describe the same domain, and the
#     versions in requirements/requirements.txt are the ones actually loaded;
#   * an object moves transient -> pending -> persistent -> detached, and each
#     transition is observed rather than asserted from memory;
#   * flush is not commit — a genuinely separate connection cannot see the
#     flushed row until the transaction commits;
#   * the N+1 problem really is 1 + N, and selectinload really is exactly 2
#     statements while joinedload really is exactly 1;
#   * DetachedInstanceError is provoked twice, on a column and on a
#     relationship, and fixed two different ways;
#   * a flush inside a loop costs 500 cursor executions where one batched
#     flush costs 1 — and, honestly, a batched ORM insert costs the SAME
#     number of executions as Core, which is not what folklore says;
#   * the lab leaves no database, no __pycache__ and no temporary directory
#     behind, opens no socket, and contains no URL and no sudo.
#
# Everything runs offline against in-memory and temporary SQLite databases.
# Deterministic, non-interactive, exits 0 only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

# Bytecode left by an EARLIER command is not this run's litter. The README
# documents `pytest starter -q`, and running it writes .pyc files that would
# then fail the cleanliness check at the end of this script -- failing the
# reader for following the instructions. Clearing them here makes that final
# check measure what it claims to: what THIS run left behind. `.venv` is
# untouched, because the packages' own bytecode is theirs, not ours.
find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -exec rm -rf {} + 2>/dev/null || true
find "${lab_dir}" -name '.venv' -prune -o -type d -name '.pytest_cache' -exec rm -rf {} + 2>/dev/null || true

failures=0
checks=0

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

check_eq() {
  local label="$1" want="$2" got="$3"
  checks=$((checks + 1))
  if [ "${want}" = "${got}" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        expected: ${want}"
    echo "        actual  : ${got}"
    failures=$((failures + 1))
  fi
}

# Assert a line matching the pattern exists in the captured file.
check_grep() {
  local label="$1" file="$2" pattern="$3"
  checks=$((checks + 1))
  if grep -qE "${pattern}" "${file}"; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        no line in $(basename "${file}") matched: ${pattern}"
    failures=$((failures + 1))
  fi
}

# Resolve a tool: an explicit override, then this lab's .venv, then whatever is
# on PATH. Fails loudly with install instructions rather than skipping quietly.
resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if [ -x "${lab_dir}/.venv/bin/${tool}" ]; then echo "${lab_dir}/.venv/bin/${tool}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

install_hint() {
  echo "  Install the pinned dependency with:" >&2
  echo "    cd ${lab_dir}" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  echo "  Or point this suite at an existing interpreter:" >&2
  echo "    PYTHON=/path/to/python3 PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
}

python_bin="$(resolve_tool python3 "${PYTHON:-}")" || {
  echo "FAIL: python3 not found." >&2
  install_hint
  exit 1
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  install_hint
  exit 1
}

if ! "${python_bin}" -c "import sqlalchemy" >/dev/null 2>&1; then
  echo "FAIL: SQLAlchemy is not importable from ${python_bin}." >&2
  echo "  This lab is about SQLAlchemy, so there is nothing to fall back to." >&2
  install_hint
  exit 1
fi

work="$(mktemp -d)"
cleanup() { rm -rf "${work}"; }
trap cleanup EXIT

export PYTHONPATH="${lab_dir}/examples"

echo "Day 093 — ORMs and SQLAlchemy"
echo

# ---------------------------------------------------------------------------
echo "1. Environment — the versions actually in use"
# ---------------------------------------------------------------------------
"${python_bin}" - > "${work}/versions.txt" <<'PY'
import platform
import sqlalchemy
import sqlite3
print(f"python {platform.python_version()}")
print(f"sqlalchemy {sqlalchemy.__version__}")
print(f"sqlite {sqlite3.sqlite_version}")
PY
sed 's/^/    /' "${work}/versions.txt"

pinned="$(grep -iE '^SQLAlchemy==' "${lab_dir}/requirements/requirements.txt" | cut -d= -f3)"
installed="$(awk '/^sqlalchemy /{print $2}' "${work}/versions.txt")"
check_eq "the installed SQLAlchemy is the version requirements.txt pins" \
  "${pinned}" "${installed}"
# Written as a plain if rather than a `case` inside `$( )`: bash 3.2, which is
# still what macOS ships, mis-parses the `)` of a case pattern as the end of
# the command substitution and silently yields the wrong answer.
if [ "${installed%%.*}" = "2" ]; then major_is_two=yes; else major_is_two=no; fi
check "SQLAlchemy is 2.x, so the modern declarative API is available" \
  "${major_is_two}"

# The lesson claims Alembic is not exercised here. Prove the claim is honest
# rather than merely stated: if it were installed, the claim would need
# rewriting, so the harness would rather fail than let the text go stale.
check "the lesson's claim that Alembic is not installed here is still true" \
  "$("${python_bin}" -c "import alembic" >/dev/null 2>&1 && echo no || echo yes)"

# ---------------------------------------------------------------------------
echo
echo "2. The toy ORM — DDL, DML, mapping back, and an identity map"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_toy.py" > "${work}/toy.txt" 2>&1
check_eq "demo_toy.py exits 0" "0" "$?"

check_grep "the class declaration generates its own CREATE TABLE" \
  "${work}/toy.txt" '^CREATE TABLE members \(id INTEGER PRIMARY KEY, name TEXT, email TEXT\)$'
check_grep "add() emits no SQL — only the two CREATE TABLEs so far" \
  "${work}/toy.txt" 'statements emitted so far: 2'
check_grep "a pending object has no primary key yet" \
  "${work}/toy.txt" 'ada\.id before flush: None'
check_grep "flush() assigns the key the database chose" \
  "${work}/toy.txt" 'ada\.id after flush: 1'
check_grep "the identity map returns the SAME object" \
  "${work}/toy.txt" 'first is second      : True'
check_grep "and it is the very object that was added" \
  "${work}/toy.txt" 'first is ada         : True'
check_grep "a repeat lookup emits zero statements" \
  "${work}/toy.txt" 'statements emitted   : 0'
check_grep "one object means one edit — no lost update" \
  "${work}/toy.txt" 'changed via .first., read via .second.: Ada O\.'

toy_statements="$(grep -cE '^     ?[0-9]+\. ' "${work}/toy.txt")"
check_eq "the toy session sent exactly 8 statements in total" "8" "${toy_statements}"

# The toy's identity map must survive a SELECT as well as a get(), which is the
# property that makes it a unit of work rather than a cache with a nice name.
"${python_bin}" - > "${work}/toy_identity.txt" <<'PY'
import sqlite3
from tiny_orm import Column, Model, Session


class Member(Model):
    __table__ = "members"
    id = Column("INTEGER", primary_key=True)
    name = Column("TEXT")


session = Session(sqlite3.connect(":memory:"))
session.create_all(Member)
ada = Member(name="Ada")
session.add(ada)
session.commit()
first = session.select(Member)[0]
second = session.get(Member, 1)
print("SAME", first is second and second is ada)
try:
    Member(nickname="oops")
    print("UNKNOWN_COLUMN rejected=False")
except TypeError:
    print("UNKNOWN_COLUMN rejected=True")
PY
check_grep "a SELECT and a get() return one object, not two copies" \
  "${work}/toy_identity.txt" '^SAME True$'
check_grep "the toy rejects a column it never declared" \
  "${work}/toy_identity.txt" '^UNKNOWN_COLUMN rejected=True$'

# ---------------------------------------------------------------------------
echo
echo "3. SQLAlchemy 2.0 — the same four operations, and the SQL it emitted"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_sqlalchemy.py" > "${work}/sqla.txt" 2>&1
check_eq "demo_sqlalchemy.py exits 0" "0" "$?"

check_grep "add() emits zero statements" \
  "${work}/sqla.txt" 'statements emitted by add\(\): 0'
check_grep "an added object is pending, not persistent" \
  "${work}/sqla.txt" 'state -> transient=False pending=True persistent=False detached=False'
check_grep "flush() emits exactly one INSERT" \
  "${work}/sqla.txt" '1\. INSERT INTO members \(name, email\) VALUES \(\?, \?\)'
check_grep "and the database assigned the key" \
  "${work}/sqla.txt" 'member\.id                  : 7'
check_grep "after the flush the object is persistent" \
  "${work}/sqla.txt" 'state -> transient=False pending=False persistent=True detached=False'
check_grep "the identity map answers a repeat get() with no SQL" \
  "${work}/sqla.txt" 'statements emitted : 0'
check_grep "select() compiles to the SQL the lesson prints" \
  "${work}/sqla.txt" 'SELECT books\.title, books\.author FROM books WHERE books\.copies >= \? ORDER BY books\.title'
check_grep "a join and an aggregate produce the expected top row" \
  "${work}/sqla.txt" 'Ada Okonkwo      3'
check_grep "the many-to-many secondary table resolves five craft books" \
  "${work}/sqla.txt" "Book\(id=8, title='Effective Java'\)"

# The declarative models must round-trip through Core metadata, because the
# claim "the ORM is built on Core" is only worth making if it is checkable.
"${python_bin}" - > "${work}/metadata.txt" <<'PY'
from models import Base
names = sorted(Base.metadata.tables)
print("TABLES", " ".join(names))
loans = Base.metadata.tables["loans"]
print("FKS", len(loans.foreign_keys))
print("SECONDARY", "book_tags" in Base.metadata.tables)
PY
check_grep "all five tables register in one MetaData" \
  "${work}/metadata.txt" '^TABLES book_tags books loans members tags$'
check_grep "the loans table carries both foreign keys" \
  "${work}/metadata.txt" '^FKS 2$'
check_grep "the many-to-many secondary is a Core Table, not a mapped class" \
  "${work}/metadata.txt" '^SECONDARY True$'

# ---------------------------------------------------------------------------
echo
echo "4. The Session as a unit of work"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_unit_of_work.py" > "${work}/uow.txt" 2>&1
check_eq "demo_unit_of_work.py exits 0" "0" "$?"

check_grep "a constructed object is transient" \
  "${work}/uow.txt" 'just constructed          -> transient'
check_grep "add() makes it pending" \
  "${work}/uow.txt" 'after session\.add\(\)       -> pending'
check_grep "flush() makes it persistent and gives it a key" \
  "${work}/uow.txt" 'after session\.flush\(\)     -> persistent   id=7'
check_grep "close() makes it detached" \
  "${work}/uow.txt" 'after session\.close\(\)     -> detached'

check_grep "the flush really did send the INSERT" \
  "${work}/uow.txt" '1\. INSERT INTO members \(name, email\) VALUES \(\?, \?\)'
check_grep "yet an outside connection still sees only 7 members" \
  "${work}/uow.txt" 'after flush, other connection sees : 7 members'
check_grep "and sees 8, with the new name, only after the commit" \
  "${work}/uow.txt" "after commit, other connection sees: 8 members, last 'Hana Ito'"

check_grep "autoflush emits the pending INSERT before an unrelated SELECT" \
  "${work}/uow.txt" '1\. INSERT INTO members \(name, email\) VALUES \(\?, \?\)'
check_grep "and the SELECT follows it in the same counted window" \
  "${work}/uow.txt" '2\. SELECT members\.id, members\.name, members\.email FROM members WHERE members\.name LIKE \?'

check_grep "reading a column after close raises DetachedInstanceError" \
  "${work}/uow.txt" 'Instance <Member at 0xADDR> is not bound to a Session; attribute refresh operation cannot proceed'
check_grep "expire_on_commit=False keeps the loaded column readable" \
  "${work}/uow.txt" "ada\.name after close: 'Ada Okonkwo'"
check_grep "touching a lazy relationship after close raises too" \
  "${work}/uow.txt" "lazy load operation of attribute 'loans' cannot proceed"
check_grep "eager loading is the fix for the relationship case" \
  "${work}/uow.txt" 'len\(ada\.loans\) after close: 4'
check_grep "the temporary database is removed on the way out" \
  "${work}/uow.txt" 'temporary database removed: True'

# ---------------------------------------------------------------------------
echo
echo "5. The N+1 problem, counted and then fixed"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_n_plus_one.py" > "${work}/nplus1.txt" 2>&1
check_eq "demo_n_plus_one.py exits 0" "0" "$?"

check_grep "the naive loop reaches all 24 loans" \
  "${work}/nplus1.txt" 'members: 6   loans reached: 24'
check_grep "lazy loading costs 1 + N = 7 statements" \
  "${work}/nplus1.txt" 'lazy \(default\)    7 statements   <- 1 \+ N'
check_grep "selectinload costs exactly 2, whatever N is" \
  "${work}/nplus1.txt" 'selectinload      2 statements   <- 1 \+ 1, whatever N is'
check_grep "joinedload costs exactly 1, at the price of wider rows" \
  "${work}/nplus1.txt" 'joinedload        1 statement    <- 1, but wider rows'
check_grep "a joinedload of a collection without unique() raises, and says why" \
  "${work}/nplus1.txt" 'The unique\(\) method must be invoked on this Result'
check_grep "the JOIN really returns 24 rows for 6 members" \
  "${work}/nplus1.txt" 'rows the JOIN actually returned : 24'
check_grep "which collapse to 6 distinct Member objects" \
  "${work}/nplus1.txt" 'distinct Member objects built   : 6'
check_grep "the identity map caps a many-to-one N+1 at the DISTINCT count" \
  "${work}/nplus1.txt" 'loans: 24   distinct titles: 8'
check_grep "joinedload flattens that many-to-one case to one statement" \
  "${work}/nplus1.txt" 'with joinedload\(Loan\.book\): 1 statement, 8 titles'
check_grep "the many-to-many is 9 statements lazily" \
  "${work}/nplus1.txt" 'lazy         9 statements'
check_grep "and 2 with selectinload" \
  "${work}/nplus1.txt" 'selectinload 2 statements'

# The counts above are the specific numbers for this seed. The property that
# matters is more general than any of them, so assert the property directly:
# eager loading must be constant in N while lazy loading grows with it.
"${python_bin}" - > "${work}/scaling.txt" <<'PY'
from sqlalchemy import insert, select
from sqlalchemy.orm import Session, selectinload

from counting import QueryCounter
from library import build_engine
from models import Member

for extra in (0, 30):
    engine = build_engine()
    if extra:
        with engine.begin() as connection:
            connection.execute(
                insert(Member),
                [
                    {"name": f"Extra {n}", "email": f"extra{n}@library.test"}
                    for n in range(extra)
                ],
            )
    with Session(engine) as session:
        with QueryCounter(engine) as lazy:
            for member in session.scalars(select(Member)).all():
                len(member.loans)
    with Session(engine) as session:
        with QueryCounter(engine) as eager:
            for member in session.scalars(
                select(Member).options(selectinload(Member.loans))
            ).all():
                len(member.loans)
    print(f"MEMBERS {6 + extra} LAZY {len(lazy)} EAGER {len(eager)}")
    engine.dispose()
PY
check_grep "with 6 members: lazy 7, eager 2" \
  "${work}/scaling.txt" '^MEMBERS 6 LAZY 7 EAGER 2$'
check_grep "with 36 members: lazy 37, eager still 2 — N+1 against a constant" \
  "${work}/scaling.txt" '^MEMBERS 36 LAZY 37 EAGER 2$'

# ---------------------------------------------------------------------------
echo
echo "6. Bulk work — and an honest reading of the numbers"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_bulk.py" > "${work}/bulk.txt" 2>&1
check_eq "demo_bulk.py exits 0" "0" "$?"

check_grep "a flush inside the loop costs one execution per row" \
  "${work}/bulk.txt" '500 cursor execution\(s\), 500 parameter set\(s\), 0 of them batched'
check_grep "add_all() with one flush batches 500 rows into 1 execution" \
  "${work}/bulk.txt" 'add_all\(\) \+ one flush        1 execution\(s\)    500 row\(s\)'
check_grep "Core insert() costs the SAME 1 execution - not fewer" \
  "${work}/bulk.txt" 'Core insert\(\), one call      1 execution\(s\)    500 row\(s\)'
check_grep "the ORM update loop must SELECT before it can UPDATE" \
  "${work}/bulk.txt" 'ORM, object by object : 2 cursor execution\(s\), 14 parameter set\(s\)'
check_grep "and it builds one object per matching row" \
  "${work}/bulk.txt" '13 Loan objects built in memory'
check_grep "a Core UPDATE is 1 execution and builds no objects" \
  "${work}/bulk.txt" 'Core UPDATE           : 1 cursor execution\(s\), 1 parameter set\(s\)'
check_grep "the object count is what scales — 1013 of them at 1000 more rows" \
  "${work}/bulk.txt" '1013 Loan objects built in memory'
check_grep "while Core stays at zero objects for the same 1013 rows" \
  "${work}/bulk.txt" '0 Loan objects built, 1013 rows changed'

# ---------------------------------------------------------------------------
echo
echo "7. The starter exercises"
# ---------------------------------------------------------------------------
(cd "${lab_dir}" && "${pytest_bin}" starter -q > "${work}/starter.txt" 2>&1)
starter_status=$?
check_eq "pytest starter exits 0 on the unmodified skeleton" "0" "${starter_status}"
check_grep "one baseline test passes and nine exercises wait" \
  "${work}/starter.txt" '1 passed, 9 skipped'

# Every skipped test must name the exercise that unblocks it, or the starter is
# a wall rather than a ladder.
skip_reasons="$(grep -c 'reason="Exercise' "${lab_dir}/starter/test_queries.py")"
check_eq "all nine skipped tests name their exercise" "9" "${skip_reasons}"
exercise_markers="$(grep -c '^    EXERCISE [0-9]' "${lab_dir}/starter/queries.py")"
check_eq "and queries.py carries a matching numbered exercise for each" \
  "10" "${exercise_markers}"

# The skeleton must actually RUN — a starter that raises before the learner has
# touched it teaches nothing except that the lab is broken.
"${python_bin}" - > "${work}/skeleton.txt" 2>&1 <<'PY'
import sys
sys.path.insert(0, "starter")
from sqlalchemy.orm import Session

from library import build_engine
from queries import books_with_at_least, loan_titles, member_loan_totals, open_loan_counts

engine = build_engine()
with Session(engine) as session:
    print("BOOKS", books_with_at_least(session, 3))
    print("COUNTS", open_loan_counts(session)[0])
    print("TOTALS", member_loan_totals(session)[0])
    print("TITLES", len(loan_titles(session)))
engine.dispose()
PY
check_grep "the unmodified skeleton returns the RIGHT answers, slowly" \
  "${work}/skeleton.txt" "^BOOKS \['Clean Code', 'Introduction to Algorithms', 'The C Programming Language'\]$"
check_grep "including the busiest borrower" \
  "${work}/skeleton.txt" "^COUNTS \('Ada Okonkwo', 3\)$"
check_grep "and every loan title" \
  "${work}/skeleton.txt" '^TITLES 24$'

# ---------------------------------------------------------------------------
echo
echo "8. Captured output still matches a live run"
# ---------------------------------------------------------------------------
for capture in toy sqlalchemy unit-of-work n-plus-one bulk; do
  case "${capture}" in
    toy)          live="${work}/toy.txt" ;;
    sqlalchemy)   live="${work}/sqla.txt" ;;
    unit-of-work) live="${work}/uow.txt" ;;
    n-plus-one)   live="${work}/nplus1.txt" ;;
    bulk)         live="${work}/bulk.txt" ;;
  esac
  stored="${lab_dir}/expected-output/${capture}.txt"
  checks=$((checks + 1))
  if [ ! -f "${stored}" ]; then
    echo "  FAIL: expected-output/${capture}.txt is missing"
    failures=$((failures + 1))
  elif diff -q "${stored}" "${live}" >/dev/null 2>&1; then
    echo "  ok: expected-output/${capture}.txt matches this run exactly"
  else
    echo "  FAIL: expected-output/${capture}.txt differs from this run"
    diff "${stored}" "${live}" | head -12 | sed 's/^/        /'
    failures=$((failures + 1))
  fi
done

# ---------------------------------------------------------------------------
echo
echo "9. Hygiene — offline, self-contained, and leaving nothing behind"
# ---------------------------------------------------------------------------
"${python_bin}" - "${lab_dir}" > "${work}/hygiene.txt" <<'PY'
import re
import sys
from pathlib import Path

root = Path(sys.argv[1])
skip = {".venv", "__pycache__", ".pytest_cache"}
urls, sudo_lines, net = set(), [], []
comment = re.compile(r"^\s*(#|--)")
for path in sorted(root.rglob("*")):
    if not path.is_file() or path.suffix not in {".py", ".sh", ".ini"}:
        continue
    if skip & set(path.parts):
        continue
    for number, line in enumerate(
        path.read_text(encoding="utf-8", errors="ignore").splitlines(), 1
    ):
        urls.update(re.findall(r"https?://[^\s\"')]+", line))
        if re.search(r"(^|[;|&(]\s*)sudo\s", line) and not comment.match(line):
            sudo_lines.append(f"{path.name}:{number}")
        if re.match(r"\s*(import|from)\s+(urllib|http|requests)\b", line):
            net.append(f"{path.name}:{number}")
print("URLS " + " ".join(sorted(urls)))
print("SUDO " + " ".join(sudo_lines))
print("NET " + " ".join(net))
PY
check_eq "no URL appears anywhere in the lab's scripts" "URLS" \
  "$(grep '^URLS ' "${work}/hygiene.txt" | sed 's/ *$//')"
check_eq "no line in this lab would actually invoke sudo" "SUDO" \
  "$(grep '^SUDO ' "${work}/hygiene.txt" | sed 's/ *$//')"
check_eq "nothing here imports an HTTP client" "NET" \
  "$(grep '^NET ' "${work}/hygiene.txt" | sed 's/ *$//')"

check "the starter test suite arms a guard against opening a socket" \
  "$(grep -q 'NetworkAccessAttempted' "${lab_dir}/starter/conftest.py" && echo yes || echo no)"

# And prove that guard is not decorative, by tripping it on purpose.
(cd "${lab_dir}/starter" && "${python_bin}" - > "${work}/guard.txt" 2>&1 <<'PY'
import conftest  # noqa: F401  — importing it arms the guard
import socket

try:
    socket.create_connection(("127.0.0.1", 9))
    print("GUARD armed=False")
except conftest.NetworkAccessAttempted:
    print("GUARD armed=True")
PY
)
check_grep "and the guard really refuses a connection when one is attempted" \
  "${work}/guard.txt" '^GUARD armed=True$'

check "no captured output leaks an absolute home path" \
  "$(grep -rl '/Users/\|/home/' "${lab_dir}/expected-output" >/dev/null 2>&1 && echo no || echo yes)"
check "every invented email address uses the library.test domain" \
  "$(grep -ohE '[A-Za-z0-9._-]+@[A-Za-z0-9._-]+' "${lab_dir}/examples/library.py" | grep -v '@library\.test$' >/dev/null 2>&1 && echo no || echo yes)"
check "this suite created no database file inside the lab directory" \
  "$(find "${lab_dir}" -name '*.db' -not -path '*/.venv/*' | grep -q . && echo no || echo yes)"
check "and left no __pycache__ behind" \
  "$(find "${lab_dir}" -type d -name '__pycache__' -not -path '*/.venv/*' | grep -q . && echo no || echo yes)"

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]