Hands-on lab — Day 90: SQLite from Python

"""executemany against a loop, and the transaction that dwarfs both.

The point of this script is not a benchmark figure — those move with the
machine, the filesystem and what else is running. The point is the SHAPE:
which of three choices actually matters, and by roughly how much.

Run it:  python3 bulk_insert.py [row_count]
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
import time
from contextlib import closing
from pathlib import Path

from db import transaction

CREATE = "CREATE TABLE rows_in (n INTEGER NOT NULL, payload TEXT NOT NULL) STRICT"
INSERT = "INSERT INTO rows_in (n, payload) VALUES (?, ?)"


def make_rows(count: int) -> list[tuple[int, str]]:
    return [(n, f"payload-for-row-{n}") for n in range(count)]


def timed(label: str, path: Path, rows: list[tuple[int, str]], body) -> float:
    with closing(sqlite3.connect(path, isolation_level=None)) as connection:
        connection.execute(CREATE)
        start = time.perf_counter()
        body(connection, rows)
        elapsed = time.perf_counter() - start
        stored = connection.execute("SELECT count(*) FROM rows_in").fetchone()[0]
    assert stored == len(rows), f"{label}: stored {stored}, expected {len(rows)}"
    return elapsed


def loop_no_transaction(connection: sqlite3.Connection, rows) -> None:
    """One statement, one transaction, one fsync — per row. The slow one."""
    for row in rows:
        connection.execute(INSERT, row)


def loop_in_transaction(connection: sqlite3.Connection, rows) -> None:
    """The same loop, wrapped once. This is where nearly all the win is."""
    with transaction(connection):
        for row in rows:
            connection.execute(INSERT, row)


def execute_many(connection: sqlite3.Connection, rows) -> None:
    """One prepared statement, stepped once per row, inside one transaction."""
    with transaction(connection):
        connection.executemany(INSERT, rows)


def main() -> int:
    count = int(sys.argv[1]) if len(sys.argv) > 1 else 20_000
    rows = make_rows(count)
    sandbox = Path(tempfile.mkdtemp(prefix="day090-bulk-"))
    try:
        print(f"Inserting {count:,} rows, three ways, into a fresh file each time.")
        print()
        results = {
            "a loop, no transaction": timed(
                "loop", sandbox / "a.db", rows, loop_no_transaction),
            "a loop inside one transaction": timed(
                "loop-txn", sandbox / "b.db", rows, loop_in_transaction),
            "executemany inside one transaction": timed(
                "many", sandbox / "c.db", rows, execute_many),
        }
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)

    slowest = max(results.values())
    width = max(len(label) for label in results)
    print(f"{'method':<{width}}  {'seconds':>9}  {'rows/second':>13}  relative")
    print(f"{'-' * width}  {'-' * 9}  {'-' * 13}  {'-' * 8}")
    for label, elapsed in results.items():
        print(f"{label:<{width}}  {elapsed:>9.4f}  {count / elapsed:>13,.0f}  "
              f"{slowest / elapsed:>6.1f}x")

    print()
    print("Read it in this order:")
    print("  * The first row pays a COMMIT — and on a durable filesystem an")
    print("    fsync — once per row. That is the cost that dominates, and it")
    print("    is the reason a bulk load in a loop feels broken.")
    print("  * Wrapping the same loop in one transaction removes almost all of")
    print("    it. Batching your writes matters more than which method you use.")
    print("  * executemany then saves the remaining per-row cost of going")
    print("    through the module: it prepares the statement once and steps it")
    print("    once per row, binding new values each time.")
    print()
    print("These numbers are from one run on one machine and will differ on")
    print("yours. The ORDER is the durable fact, not the figures.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Cursors, the four ways to get rows out, and row factories.

Run it:  python3 cursors_demo.py

Everything happens in a database built inside a temporary directory, which
is removed on the way out.
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
import tracemalloc
from contextlib import closing
from pathlib import Path

from db import BookRepository, apply_schema, connect, dict_factory, transaction
from domain import Book


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


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-cursors-"))
    try:
        return run(sandbox)
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)


def run(sandbox: Path) -> int:
    import seed

    path = sandbox / "library.db"

    # `closing` is the idiom for "close this when the block ends". Note that
    # `with connection:` would NOT close it — that form manages a
    # transaction, not the connection's lifetime.
    with closing(seed.build(path)) as connection:
        books = BookRepository(connection)

        rule("1. execute returns a cursor, and the cursor is the result")
        cursor = connection.execute("SELECT book_id, title, year FROM books ORDER BY year")
        print(f"    type(connection.execute(...)) -> {type(cursor).__name__}")
        print(f"    cursor.description names the columns: "
              f"{[column[0] for column in cursor.description]}")
        print("    cursor.rowcount for a SELECT ->", cursor.rowcount,
              "(SQLite cannot know how many rows a query will yield until it has run it)")
        print("    the same cursor, reused for a second statement, forgets the first.")

        rule("2. fetchone, fetchmany, fetchall, and iteration")
        cursor = connection.execute("SELECT title, year FROM books ORDER BY year")
        first = cursor.fetchone()
        print(f"    fetchone()      -> {tuple(first)}")
        batch = cursor.fetchmany(3)
        print(f"    fetchmany(3)    -> {[row['title'] for row in batch]}")
        rest = cursor.fetchall()
        print(f"    fetchall()      -> {len(rest)} remaining rows: {[row['title'] for row in rest]}")
        print(f"    fetchone() now  -> {cursor.fetchone()}   (the cursor is exhausted)")
        print()
        print("    iterating the cursor instead, which is the memory-safe default:")
        for book in books.stream_all():
            print(f"      {book.book_id}: {book.label}")

        rule("3. what fetchall actually costs, measured")
        # 40,000 rows in a second table, so the difference is visible.
        connection.execute("CREATE TABLE wide (n INTEGER, payload TEXT) STRICT")
        with transaction(connection):
            connection.executemany(
                "INSERT INTO wide (n, payload) VALUES (?, ?)",
                [(n, "x" * 200) for n in range(40_000)],
            )

        tracemalloc.start()
        rows = connection.execute("SELECT n, payload FROM wide").fetchall()
        fetchall_peak = tracemalloc.get_traced_memory()[1]
        tracemalloc.stop()
        del rows

        tracemalloc.start()
        total = 0
        for row in connection.execute("SELECT n, payload FROM wide"):
            total += row["n"]
        iterate_peak = tracemalloc.get_traced_memory()[1]
        tracemalloc.stop()

        print(f"    40,000 rows of about 200 bytes each")
        print(f"    fetchall()        peak traced memory: {fetchall_peak:>10,} bytes")
        print(f"    iterating cursor  peak traced memory: {iterate_peak:>10,} bytes")
        print(f"    ratio: {fetchall_peak / max(iterate_peak, 1):.0f}x")
        print(f"    (both computed the same sum: {total:,})")
        print("    fetchall builds the whole list first. Iteration holds one row.")

        rule("4. row factories: tuple, sqlite3.Row, and a dict")
        plain = sqlite3.connect(path)
        row = plain.execute("SELECT title, year FROM books ORDER BY year").fetchone()
        print(f"    default (no factory) -> {row!r}   — addressed by position only")

        plain.row_factory = sqlite3.Row
        row = plain.execute("SELECT title, year FROM books ORDER BY year").fetchone()
        print(f"    sqlite3.Row          -> row['title'] = {row['title']!r}, row[0] = {row[0]!r}")
        print(f"                            row.keys() = {row.keys()}")
        print(f"                            isinstance(row, dict) = {isinstance(row, dict)}"
              "   — it is NOT a dict; it has no .get and json.dumps refuses it")
        print(f"                            dict(row) = {dict(row)}")

        plain.row_factory = dict_factory
        row = plain.execute("SELECT title, year FROM books ORDER BY year").fetchone()
        print(f"    dict_factory         -> {row}   type: {type(row).__name__}")
        plain.close()

        rule("5. a cursor of your own, when you want two open at once")
        outer = connection.cursor()
        inner = connection.cursor()
        outer.execute("SELECT book_id, title FROM books ORDER BY book_id")
        for book_row in outer.fetchmany(2):
            inner.execute(
                "SELECT count(*) AS n FROM loans WHERE book_id = ?", (book_row["book_id"],)
            )
            print(f"    {book_row['title']:<36} loans: {inner.fetchone()['n']}")
        print("    Two cursors on one connection: independent positions, one transaction.")

    print()
    print("connection closed by contextlib.closing; sandbox removed.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""The data layer, built from first principles.

Five pieces, and nothing else:

  1. `SCHEMA`             — the tables, as one script.
  2. `connect()`          — a connection factory that sets the four things
                            every connection in this program must have.
  3. `transaction()`      — a context manager that begins, commits, and rolls
                            back explicitly.
  4. `row_to_book()`      — one function that turns a database row into a
                            domain object, so mapping lives in one place.
  5. `BookRepository`     — every SQL statement in the program, and nothing
                            but SQL statements.

The rule this file exists to enforce: **SQL lives here and only here.** No
other module in the lab imports `sqlite3`. Grep for it and see —
`tests/run_tests.sh` does exactly that and fails if it finds one.

Every statement below is parameterised. There is not one f-string, one `%`
and one `+` anywhere near a SQL string in this file, and the test suite
proves that mechanically rather than trusting the author.
"""

from __future__ import annotations

import sqlite3
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from pathlib import Path

from domain import Book, BookNotFound, DuplicateTitle, Loan, Member

SCHEMA = """
CREATE TABLE IF NOT EXISTS books (
    book_id  INTEGER PRIMARY KEY,
    title    TEXT    NOT NULL UNIQUE,
    author   TEXT    NOT NULL,
    year     INTEGER NOT NULL CHECK (year BETWEEN 1400 AND 2100),
    copies   INTEGER NOT NULL CHECK (copies >= 0)
) STRICT;

CREATE TABLE IF NOT EXISTS members (
    member_id INTEGER PRIMARY KEY,
    name      TEXT NOT NULL,
    email     TEXT NOT NULL UNIQUE
) STRICT;

CREATE TABLE IF NOT EXISTS loans (
    loan_id     INTEGER PRIMARY KEY,
    book_id     INTEGER NOT NULL REFERENCES books(book_id),
    member_id   INTEGER NOT NULL REFERENCES members(member_id),
    borrowed_on TEXT NOT NULL,
    due_on      TEXT NOT NULL,
    returned_on TEXT,
    CHECK (due_on >= borrowed_on)
) STRICT;

CREATE INDEX IF NOT EXISTS loans_open ON loans(due_on) WHERE returned_on IS NULL;
"""


# ---------------------------------------------------------------------------
# 1. The connection factory
# ---------------------------------------------------------------------------

def connect(path: str | Path) -> sqlite3.Connection:
    """Open a connection configured the way this program needs it.

    Four decisions, each of which is per-connection and would otherwise have
    to be remembered at every call site:

    * `isolation_level=None` — turn OFF the module's implicit transaction
      handling. Nothing begins a transaction behind our back; `transaction()`
      below is the only thing that opens one, which is what makes the
      transaction boundaries visible in the code rather than implied.
    * `PRAGMA foreign_keys = ON` — foreign keys are OFF by default in SQLite,
      per connection. Without this line every REFERENCES clause in SCHEMA is
      a comment. It is executed here, outside any transaction, because the
      pragma is a silent no-op inside one.
    * `row_factory = sqlite3.Row` — rows arrive addressable by column name
      instead of by position, so `row["title"]` survives a change to the
      SELECT list that `row[1]` would not.
    * a busy timeout — if another connection holds the write lock, wait
      rather than raising immediately.

    The caller owns the connection and must close it. `contextlib.closing`
    is the tidy way; see `report.py`.
    """
    connection = sqlite3.connect(str(path), isolation_level=None, timeout=5.0)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    return connection


def dict_factory(cursor: sqlite3.Cursor, row: tuple) -> dict:
    """An alternative row factory: plain dicts instead of sqlite3.Row.

    `sqlite3.Row` is not a dict — it has no `.get`, it is immutable, and
    `json.dumps` refuses it. When a row has to leave the data layer as JSON,
    this is the three-line answer. `cursor.description` is a 7-tuple per
    column of which only the first item, the name, is populated by sqlite3.
    """
    return {column[0]: value for column, value in zip(cursor.description, row)}


def apply_schema(connection: sqlite3.Connection) -> None:
    """Create the tables.

    `executescript` is the right tool for a multi-statement script and the
    wrong tool for anything containing a value: it takes no parameters at
    all, so anything variable would have to be pasted into the string. It
    also issues an implicit COMMIT before it runs, which means it cannot be
    nested inside `transaction()`.
    """
    connection.executescript(SCHEMA)


# ---------------------------------------------------------------------------
# 2. The transaction context manager
# ---------------------------------------------------------------------------

@contextmanager
def transaction(connection: sqlite3.Connection) -> Iterator[sqlite3.Connection]:
    """Run a block of writes as one indivisible act.

    Commits if the block finishes, rolls back if anything at all is raised,
    and re-raises. `BaseException` rather than `Exception` on purpose: a
    KeyboardInterrupt in the middle of a two-statement change must roll back
    too, and `except Exception` would let it through with the transaction
    still open.

    This is not the same thing as `with connection:`. That form commits or
    rolls back a transaction the module opened implicitly, and it does NOT
    close the connection — a genuinely common misreading. This function is
    explicit about both ends, which is why the connection factory turns the
    implicit machinery off.
    """
    if connection.in_transaction:
        raise RuntimeError(
            "a transaction is already open on this connection; "
            "SQLite has no nested transactions, only SAVEPOINTs"
        )
    connection.execute("BEGIN")
    try:
        yield connection
    except BaseException:
        connection.rollback()
        raise
    else:
        connection.commit()


# ---------------------------------------------------------------------------
# 3. Row-to-object mapping — in one place, on purpose
# ---------------------------------------------------------------------------

def row_to_book(row: sqlite3.Row) -> Book:
    """Turn one database row into one domain object.

    Every SELECT in the repository returns these five columns in this order
    and hands the row here. When a column is renamed, this function is the
    only thing that changes.
    """
    return Book(
        book_id=row["book_id"],
        title=row["title"],
        author=row["author"],
        year=row["year"],
        copies=row["copies"],
    )


def row_to_member(row: sqlite3.Row) -> Member:
    return Member(member_id=row["member_id"], name=row["name"], email=row["email"])


def row_to_loan(row: sqlite3.Row) -> Loan:
    return Loan(
        loan_id=row["loan_id"],
        book_id=row["book_id"],
        member_id=row["member_id"],
        borrowed_on=row["borrowed_on"],
        due_on=row["due_on"],
        returned_on=row["returned_on"],
    )


# ---------------------------------------------------------------------------
# 4. The repository
# ---------------------------------------------------------------------------

# Sorting has to vary at runtime, and a placeholder cannot stand in for an
# identifier: `ORDER BY ?` binds a VALUE, and ordering every row by the same
# constant is not ordering at all. The safe form is an allow-list — and the
# safest allow-list holds whole statements you wrote, so that no SQL string
# in this file is ever built out of pieces.
SORTED_QUERIES = {
    "title": "SELECT book_id, title, author, year, copies FROM books ORDER BY title",
    "author": "SELECT book_id, title, author, year, copies FROM books ORDER BY author",
    "year": "SELECT book_id, title, author, year, copies FROM books ORDER BY year",
}


class BookRepository:
    """Every SQL statement about books, and no statement about anything else.

    The repository takes a connection rather than a path. That one decision
    is what makes it testable: the suite hands it a connection to a database
    in a temporary directory, and nothing in the class knows the difference.
    """

    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection

    # -- reads ------------------------------------------------------------

    def get(self, book_id: int) -> Book:
        row = self._connection.execute(
            "SELECT book_id, title, author, year, copies FROM books WHERE book_id = ?",
            (book_id,),
        ).fetchone()
        if row is None:
            raise BookNotFound(f"no book with id {book_id}")
        return row_to_book(row)

    def find_by_author(self, author: str) -> list[Book]:
        """The query the injection demonstration attacks.

        `author` arrives from outside this program. It is bound, so the
        engine compares it to a column and never parses it.
        """
        cursor = self._connection.execute(
            "SELECT book_id, title, author, year, copies FROM books"
            " WHERE author = ? ORDER BY year",
            (author,),
        )
        return [row_to_book(row) for row in cursor]

    def published_between(self, first: int, last: int) -> list[Book]:
        """Named placeholders, which read better as soon as there are three.

        Named style takes a mapping; qmark style takes a sequence. Mixing
        them raises sqlite3.ProgrammingError rather than doing something
        surprising.
        """
        cursor = self._connection.execute(
            "SELECT book_id, title, author, year, copies FROM books"
            " WHERE year BETWEEN :first AND :last ORDER BY year",
            {"first": first, "last": last},
        )
        return [row_to_book(row) for row in cursor]

    def all_sorted(self, sort_key: str = "title") -> list[Book]:
        """Sorting by a column chosen at runtime, done safely.

        The key selects a whole statement from SORTED_QUERIES. An unknown
        key raises before any SQL exists at all. This is the only correct
        way to vary an identifier: an allow-list you wrote, never a value
        you received — and note that no string is assembled even here.
        """
        try:
            statement = SORTED_QUERIES[sort_key]
        except KeyError:
            raise ValueError(
                f"cannot sort by {sort_key!r}; choose from {sorted(SORTED_QUERIES)}"
            ) from None
        return [row_to_book(row) for row in self._connection.execute(statement)]

    def stream_all(self) -> Iterator[Book]:
        """Iterate the cursor instead of fetching everything.

        `fetchall()` on a ten-million-row table builds a ten-million-item
        list in memory. Iterating a cursor asks the virtual machine for one
        row per step, so memory stays flat however large the table is. This
        is the default worth reaching for; `fetchall` is the special case.
        """
        cursor = self._connection.execute(
            "SELECT book_id, title, author, year, copies FROM books ORDER BY book_id"
        )
        for row in cursor:
            yield row_to_book(row)

    def count(self) -> int:
        return self._connection.execute("SELECT count(*) AS n FROM books").fetchone()["n"]

    # -- writes -----------------------------------------------------------

    def add(self, book: Book) -> Book:
        """Insert one book and return it with the id the database assigned.

        `cursor.lastrowid` is the rowid of the last successful INSERT on
        THAT cursor — which is why the cursor returned by `execute` is worth
        keeping rather than discarding.
        """
        try:
            cursor = self._connection.execute(
                "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                (book.title, book.author, book.year, book.copies),
            )
        except sqlite3.IntegrityError as error:
            # Translate the storage error into a domain error at the boundary.
            # Nothing above this line should ever have to import sqlite3 to
            # find out that a title was taken.
            if "books.title" in str(error):
                raise DuplicateTitle(f"a book titled {book.title!r} is already stored") from error
            raise
        return Book(
            book_id=cursor.lastrowid,
            title=book.title,
            author=book.author,
            year=book.year,
            copies=book.copies,
        )

    def add_many(self, books: Sequence[Book]) -> int:
        """Bulk insert with one prepared statement and many bindings.

        `executemany` compiles the statement once and steps it once per row.
        The loop version compiles it once per row as well, and pays the
        round trip through the module each time. `bulk_insert.py` measures
        the difference rather than asserting it.
        """
        cursor = self._connection.executemany(
            "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
            [(b.title, b.author, b.year, b.copies) for b in books],
        )
        return cursor.rowcount

    def set_copies(self, book_id: int, copies: int) -> None:
        cursor = self._connection.execute(
            "UPDATE books SET copies = ? WHERE book_id = ?",
            (copies, book_id),
        )
        if cursor.rowcount == 0:
            raise BookNotFound(f"no book with id {book_id}")

    def delete(self, book_id: int) -> None:
        cursor = self._connection.execute("DELETE FROM books WHERE book_id = ?", (book_id,))
        if cursor.rowcount == 0:
            raise BookNotFound(f"no book with id {book_id}")


class LoanRepository:
    """The loans half, kept separate so each class stays readable."""

    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection

    def add_member(self, member: Member) -> Member:
        cursor = self._connection.execute(
            "INSERT INTO members (name, email) VALUES (?, ?)",
            (member.name, member.email),
        )
        return Member(member_id=cursor.lastrowid, name=member.name, email=member.email)

    def borrow(self, book_id: int, member_id: int, borrowed_on: str, due_on: str) -> Loan:
        """Two writes that are only true together — the reason transactions exist.

        A loan row appears and the shelf count drops. Call this inside
        `transaction()`; if the second statement fails, the first must not
        survive, and `test_repository.py` proves that it does not.
        """
        cursor = self._connection.execute(
            "INSERT INTO loans (book_id, member_id, borrowed_on, due_on)"
            " VALUES (?, ?, ?, ?)",
            (book_id, member_id, borrowed_on, due_on),
        )
        self._connection.execute(
            "UPDATE books SET copies = copies - 1 WHERE book_id = ?",
            (book_id,),
        )
        return Loan(
            loan_id=cursor.lastrowid,
            book_id=book_id,
            member_id=member_id,
            borrowed_on=borrowed_on,
            due_on=due_on,
        )

    def overdue(self, as_of: str) -> list[dict]:
        """The three-table question, answered in one statement.

        The returned rows are dicts rather than domain objects because this
        is a report, not an entity: nothing here has an identity to preserve.
        """
        cursor = self._connection.execute(
            """
            SELECT   members.name AS borrower,
                     books.title  AS book,
                     loans.due_on AS due,
                     julianday(:as_of) - julianday(loans.due_on) AS days_late
            FROM     loans
            JOIN     books   ON books.book_id     = loans.book_id
            JOIN     members ON members.member_id = loans.member_id
            WHERE    loans.returned_on IS NULL
              AND    loans.due_on < :as_of
            ORDER BY days_late DESC
            """,
            {"as_of": as_of},
        )
        return [
            {
                "borrower": row["borrower"],
                "book": row["book"],
                "due": row["due"],
                "days_late": int(row["days_late"]),
            }
            for row in cursor
        ]

    def open_count(self) -> int:
        return self._connection.execute(
            "SELECT count(*) AS n FROM loans WHERE returned_on IS NULL"
        ).fetchone()["n"]

"""The domain objects — Day 70's model, unchanged and unaware of storage.

Read the imports at the top of this file. There is no `sqlite3` here, and
there never will be. That absence is the whole architectural claim of this
lab: the objects that carry your program's meaning do not know that a
database exists, which is exactly what makes them testable without one
(Day 74) and replaceable without rewriting them.

Everything that knows about SQL lives in `db.py`. Everything that knows
about the problem lives here.
"""

from __future__ import annotations

from dataclasses import dataclass


class LibraryError(Exception):
    """Base class for every error this domain raises.

    The outer layer catches this one class and can say something a human can
    act on, instead of leaking an `sqlite3.IntegrityError` from three layers
    down into a user interface.
    """


class InvalidBook(LibraryError):
    """A Book was asked to exist in a state the domain forbids."""


class BookNotFound(LibraryError):
    """A lookup by identity found nothing."""


class DuplicateTitle(LibraryError):
    """A write would have created a second book with an existing title."""


@dataclass(frozen=True)
class Book:
    """One book. Frozen, because a book's identity does not change.

    `book_id` is `None` for a book that has been built in memory but never
    stored. The repository fills it in when the database assigns one, by
    returning a new Book rather than mutating this one.
    """

    title: str
    author: str
    year: int
    copies: int
    book_id: int | None = None

    def __post_init__(self) -> None:
        if not self.title.strip():
            raise InvalidBook("a book must have a title")
        if not self.author.strip():
            raise InvalidBook("a book must have an author")
        if not isinstance(self.year, int) or not (1400 <= self.year <= 2100):
            raise InvalidBook(f"year out of range: {self.year!r}")
        if not isinstance(self.copies, int) or self.copies < 0:
            raise InvalidBook(f"copies must be a non-negative integer: {self.copies!r}")

    @property
    def label(self) -> str:
        """How a book prints in a report. Presentation, not persistence."""
        return f"{self.title} ({self.author}, {self.year})"


@dataclass(frozen=True)
class Member:
    """One library member."""

    name: str
    email: str
    member_id: int | None = None

    def __post_init__(self) -> None:
        if not self.name.strip():
            raise LibraryError("a member must have a name")
        if "@" not in self.email:
            raise LibraryError(f"not an address: {self.email!r}")


@dataclass(frozen=True)
class Loan:
    """One book, out with one member, due on one date.

    `returned_on` is None while the book is still out. That is the one place
    in this model where None means something specific rather than "missing".
    """

    book_id: int
    member_id: int
    borrowed_on: str
    due_on: str
    returned_on: str | None = None
    loan_id: int | None = None

    @property
    def is_open(self) -> bool:
        return self.returned_on is None

"""Which mistake raises which exception — produced by making each mistake.

The table this prints is not copied from documentation. Each row is one
deliberate error, caught, and reported with the class the interpreter
actually raised.

Run it:  python3 errors_demo.py
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
from contextlib import closing
from pathlib import Path

from db import apply_schema, connect
from domain import Book

SETUP = [
    ("INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
     ("The Mythical Man-Month", "Fred Brooks", 1975, 1)),
    ("INSERT INTO members (name, email) VALUES (?, ?)",
     ("Ada Lovelace", "ada@example.invalid")),
]


def attempt(connection: sqlite3.Connection, label: str, call) -> tuple[str, str, str]:
    """Run one deliberate mistake and report what came back."""
    try:
        call()
    except sqlite3.Error as error:
        return (label, type(error).__name__, str(error))
    except Exception as error:  # noqa: BLE001 - we want to see anything else too
        return (label, type(error).__name__ + " (not a sqlite3.Error)", str(error))
    return (label, "nothing raised", "")


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-errors-"))
    try:
        return run(sandbox)
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)


def run(sandbox: Path) -> int:
    print("The sqlite3 exception hierarchy, as this interpreter defines it:")
    for cls in (
        sqlite3.Warning,
        sqlite3.Error,
        sqlite3.InterfaceError,
        sqlite3.DatabaseError,
        sqlite3.DataError,
        sqlite3.OperationalError,
        sqlite3.IntegrityError,
        sqlite3.InternalError,
        sqlite3.ProgrammingError,
        sqlite3.NotSupportedError,
    ):
        parents = " <- ".join(base.__name__ for base in cls.__mro__[1:-1])
        print(f"    sqlite3.{cls.__name__:<18} {parents}")
    print()
    print("    Everything except Warning descends from sqlite3.Error, which is")
    print("    the one class a data layer should catch at its outer edge.")
    print()

    with closing(connect(sandbox / "errors.db")) as connection:
        apply_schema(connection)
        for statement, values in SETUP:
            connection.execute(statement, values)
        connection.commit()

        results = [
            attempt(connection, "duplicate title (UNIQUE)",
                    lambda: connection.execute(
                        "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                        ("The Mythical Man-Month", "Fred Brooks", 1975, 1))),
            attempt(connection, "missing required column (NOT NULL)",
                    lambda: connection.execute(
                        "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                        (None, "Fred Brooks", 1975, 1))),
            attempt(connection, "value fails a CHECK",
                    lambda: connection.execute(
                        "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                        ("Negative Copies", "Nobody", 1975, -1))),
            attempt(connection, "loan naming a member who does not exist (FOREIGN KEY)",
                    lambda: connection.execute(
                        "INSERT INTO loans (book_id, member_id, borrowed_on, due_on)"
                        " VALUES (?, ?, ?, ?)",
                        (1, 999, "2026-08-01", "2026-08-15"))),
            attempt(connection, "wrong type into a STRICT column",
                    lambda: connection.execute(
                        "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                        ("Bad Year", "Nobody", "nineteen seventy", 1))),
            attempt(connection, "table that is not there",
                    lambda: connection.execute("SELECT * FROM shelves")),
            attempt(connection, "column that is not there",
                    lambda: connection.execute("SELECT isbn FROM books")),
            attempt(connection, "SQL that is not SQL",
                    lambda: connection.execute("SELEKT * FROM books")),
            attempt(connection, "too many bindings for the placeholders",
                    lambda: connection.execute(
                        "SELECT * FROM books WHERE book_id = ?", (1, 2))),
            attempt(connection, "named placeholders given a sequence",
                    lambda: connection.execute(
                        "SELECT * FROM books WHERE book_id = :book_id", (1,))),
            attempt(connection, "two statements in one execute()",
                    lambda: connection.execute("SELECT 1; SELECT 2;")),
            attempt(connection, "binding a Python type SQLite has no column for",
                    lambda: connection.execute(
                        "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                        ({"title": "a dict"}, "Nobody", 1975, 1))),
            attempt(connection, "using a connection after close()",
                    lambda: use_closed(sandbox)),
        ]

    width_label = max(len(row[0]) for row in results)
    width_class = max(len(row[1]) for row in results)
    print(f"{'the mistake':<{width_label}}  {'raises':<{width_class}}  message")
    print(f"{'-' * width_label}  {'-' * width_class}  {'-' * 46}")
    for label, cls, message in results:
        print(f"{label:<{width_label}}  {cls:<{width_class}}  {message}")

    print()
    print("Read the pattern rather than the rows:")
    print("  IntegrityError    — the DATA broke a rule you wrote in the schema.")
    print("  OperationalError  — the DATABASE could not do it: no such table,")
    print("                      bad syntax, file locked, disk full.")
    print("  ProgrammingError  — YOUR CODE misused the module: wrong number of")
    print("                      bindings, two statements, a closed connection.")
    print("  The first is a fact about the user's input. The third is a bug.")

    failures = [row for row in results if row[1] == "nothing raised"]
    print()
    print(f"{len(results)} deliberate mistakes, {len(failures)} of which raised nothing.")
    return 1 if failures else 0


def use_closed(sandbox: Path) -> None:
    connection = sqlite3.connect(sandbox / "closed.db")
    connection.close()
    connection.execute("SELECT 1")


if __name__ == "__main__":
    sys.exit(main())

"""SQL injection, demonstrated rather than warned about.

This script builds a throwaway database in a fresh temporary directory,
attacks it for real, and deletes it on the way out. Nothing outside that
directory is touched, and the directory is removed even if a check fails.

Four acts:

  1. A query built by string concatenation is broken by a crafted input,
     and every private address in the table comes back.
  2. The same trick aimed at destroying a table meets `execute`, which
     refuses more than one statement — an honest limit worth knowing, and
     not a defence you may rely on.
  3. The same string handed to `executescript`, which does accept several
     statements, destroys the table for real. The damage is printed.
  4. The identical hostile strings, bound as parameters, are treated as
     ordinary text: zero rows, nothing dropped, no error.

Run it:  python3 injection_demo.py
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path

# The two inputs. Imagine each arriving from a search box, a query string,
# a CSV column, or the output of a language model asked to name an author.
LEAK = "Ada' OR '1'='1"
DESTROY = "Ada'; DROP TABLE members; --"

SCHEMA = """
CREATE TABLE members (
    member_id INTEGER PRIMARY KEY,
    name      TEXT NOT NULL,
    email     TEXT NOT NULL,
    pin       TEXT NOT NULL
);
INSERT INTO members (name, email, pin) VALUES
    ('Ada Lovelace',  'ada@example.invalid',   '4417'),
    ('Grace Hopper',  'grace@example.invalid', '9021'),
    ('Alan Turing',   'alan@example.invalid',  '1912');
"""

failures = 0


def check(label: str, ok: bool) -> None:
    global failures
    if ok:
        print(f"  ok: {label}")
    else:
        print(f"  FAIL: {label}")
        failures += 1


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


def members_table_exists(connection: sqlite3.Connection) -> bool:
    row = connection.execute(
        "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'members'"
    ).fetchone()
    return row is not None


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-injection-"))
    print(f"sandbox: a throwaway database inside {sandbox.name}/ — deleted on exit")
    try:
        return run(sandbox)
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)
        print()
        print("sandbox removed. Nothing outside it was ever opened.")


def run(sandbox: Path) -> int:
    path = sandbox / "victim.db"

    # ---------------------------------------------------------------- act 1
    rule("ACT 1 — the string-built query, and what the crafted value does to it")
    connection = sqlite3.connect(path)
    connection.executescript(SCHEMA)

    honest = "Ada Lovelace"
    print(f'  the ordinary case, input = "{honest}"')
    built = "SELECT name, email, pin FROM members WHERE name = '" + honest + "'"
    print(f"    statement: {built}")
    rows = connection.execute(built).fetchall()
    print(f"    rows: {len(rows)}  -> {rows}")
    check("the concatenated query looks perfectly fine on ordinary input", len(rows) == 1)

    print()
    print(f'  the crafted case, input = "{LEAK}"')
    built = "SELECT name, email, pin FROM members WHERE name = '" + LEAK + "'"
    print(f"    statement: {built}")
    print("    " + " " * (len("statement: ") + built.index("' OR")) + "^ the apostrophe inside the value closed the string early")
    rows = connection.execute(built).fetchall()
    print(f"    rows: {len(rows)}  -> every member, with address and PIN:")
    for row in rows:
        print(f"      {row}")
    check(
        "one apostrophe changed the statement's meaning: the WHERE clause "
        "became name = 'Ada' OR '1'='1', which is true for every row",
        len(rows) == 3,
    )
    print(
        "    Nothing was escaped, nothing was 'hacked'. The value became CODE\n"
        "    because it was inside the string before the parser ever saw it."
    )

    # ---------------------------------------------------------------- act 2
    rule("ACT 2 — the destructive version meets execute(), which takes one statement")
    built = "SELECT name FROM members WHERE name = '" + DESTROY + "'"
    print(f"    statement: {built}")
    try:
        connection.execute(built).fetchall()
        raised = "nothing"
    except sqlite3.Error as error:
        raised = f"{type(error).__name__}: {error}"
    print(f"    raised: {raised}")
    check(
        "sqlite3.Connection.execute refuses more than one statement, so this "
        "particular attack fails here",
        raised.startswith("ProgrammingError"),
    )
    check("the members table is still standing after act 2", members_table_exists(connection))
    print(
        "    Read that carefully. The attack failed because of a limit in the\n"
        "    Python module, NOT because the code was safe. Act 1 already leaked\n"
        "    every row through the same hole, and act 3 removes the limit."
    )

    # ---------------------------------------------------------------- act 3
    rule("ACT 3 — the same string handed to executescript(), which accepts many")
    print(f"    statement: {built}")
    before = connection.execute("SELECT count(*) FROM members").fetchone()[0]
    print(f"    members before: {before} rows")
    connection.executescript(built)
    exists = members_table_exists(connection)
    print(f"    members table exists afterwards: {exists}")
    try:
        connection.execute("SELECT count(*) FROM members").fetchone()
        aftermath = "still queryable"
    except sqlite3.Error as error:
        aftermath = f"{type(error).__name__}: {error}"
    print(f"    querying it now: {aftermath}")
    check("the table was destroyed by a value that arrived as text", not exists)
    check("and the program that did it contained no DROP anywhere", "DROP" not in SCHEMA)
    connection.close()

    # ---------------------------------------------------------------- act 4
    rule("ACT 4 — the identical inputs, bound as parameters")
    path2 = sandbox / "protected.db"
    connection = sqlite3.connect(path2)
    connection.executescript(SCHEMA)

    for label, value in (("leak attempt", LEAK), ("destroy attempt", DESTROY)):
        rows = connection.execute(
            "SELECT name, email, pin FROM members WHERE name = ?", (value,)
        ).fetchall()
        print(f'    {label:16} value = "{value}"')
        print(f"    {'':16} statement: SELECT name, email, pin FROM members WHERE name = ?")
        print(f"    {'':16} rows returned: {len(rows)}")
        check(f"{label}: bound as a value, it matched no member name", rows == [])

    check("the members table is untouched", members_table_exists(connection))
    check(
        "and still holds all three rows",
        connection.execute("SELECT count(*) FROM members").fetchone()[0] == 3,
    )
    print(
        "    The statement was compiled with a '?' in it BEFORE any value\n"
        "    existed. Binding cannot change a compiled statement's shape, so\n"
        "    the apostrophe is just a character in a string that no member is\n"
        "    called. The engine compared it and moved on."
    )
    connection.close()

    rule("SUMMARY")
    print("    concatenated + crafted input  -> 3 private rows leaked, then a table dropped")
    print("    parameterised + same input    -> 0 rows, no error, nothing changed")
    print("    the difference is one character: ? instead of an f-string.")
    print()
    print(f"{failures} failure(s) in this demonstration.")
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())

"""A guard: no SQL statement in this lab is built out of pieces.

Reviewing for injection by reading is unreliable — the dangerous line looks
exactly like the safe one until you notice the `f`. So this check is
mechanical, it runs in the test suite, and it fails the build.

It parses each file with `ast` rather than grepping, because a regular
expression over source text cannot tell a statement from the word "select"
in a comment or a print. Two places are inspected, and only those two:

  * the first argument of any `.execute`, `.executemany` or `.executescript`
    call — the string that actually reaches the engine;
  * any assignment to a name that says it holds SQL (`sql`, `query`,
    `statement`, `stmt`, or those names with a prefix or suffix).

In either place, four shapes are refused:

    f"SELECT ... {value} ..."        an f-string with a substitution
    "SELECT ... " + value            a concatenation with a non-literal
    "SELECT ... %s" % value          percent formatting
    "SELECT ... {}".format(value)    str.format

Adjacent string literals — "SELECT a, b" " FROM t" — are NOT flagged.
Python joins those at compile time; no runtime value can enter, so they are
the recommended way to wrap a long statement across lines.

`injection_demo.py` is exempt by name, because building a broken query on
purpose is the entire point of that file.

Run it:  python3 no_sql_strings.py [directory]
"""

from __future__ import annotations

import ast
import sys
from pathlib import Path

EXEMPT = {"injection_demo.py", "no_sql_strings.py"}

SQL_WORDS = (
    "select ", "insert into", "update ", "delete from", "create table",
    "drop table", " from ", " where ", "order by", "values (",
)


def looks_like_sql(text: str) -> bool:
    lowered = text.lower()
    return any(word in lowered for word in SQL_WORDS)


def literal_parts(node: ast.AST) -> str:
    """The constant text of an expression, ignoring the variable parts."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    if isinstance(node, ast.JoinedStr):
        return "".join(
            part.value for part in node.values
            if isinstance(part, ast.Constant) and isinstance(part.value, str)
        )
    if isinstance(node, ast.BinOp):
        return literal_parts(node.left) + literal_parts(node.right)
    return ""


EXECUTE_METHODS = {"execute", "executemany", "executescript"}
SQL_NAME_PARTS = ("sql", "query", "statement", "stmt")


def names_sql(name: str) -> bool:
    return any(part in name.lower() for part in SQL_NAME_PARTS)


class Scanner(ast.NodeVisitor):
    """Look only where a built string could actually become a statement."""

    def __init__(self, path: Path) -> None:
        self.path = path
        self.findings: list[str] = []

    def report(self, node: ast.AST, how: str, where: str) -> None:
        self.findings.append(
            f"{self.path.name}:{node.lineno}: SQL built by {how} and {where}"
        )

    def inspect(self, node: ast.AST, where: str) -> None:
        """Refuse any assembled string that reads like SQL."""
        if isinstance(node, ast.JoinedStr):
            substituted = any(isinstance(part, ast.FormattedValue) for part in node.values)
            if substituted and looks_like_sql(literal_parts(node)):
                self.report(node, "an f-string", where)
        elif isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
            both_literal = isinstance(node.left, ast.Constant) and isinstance(
                node.right, ast.Constant
            )
            if not both_literal and looks_like_sql(literal_parts(node)):
                self.report(node, "concatenation with +", where)
        elif isinstance(node, ast.BinOp) and isinstance(node.op, ast.Mod):
            if looks_like_sql(literal_parts(node.left)):
                self.report(node, "percent formatting", where)
        elif (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == "format"
            and looks_like_sql(literal_parts(node.func.value))
        ):
            self.report(node, "str.format", where)

    def visit_Call(self, node: ast.Call) -> None:
        if (
            isinstance(node.func, ast.Attribute)
            and node.func.attr in EXECUTE_METHODS
            and node.args
        ):
            self.inspect(node.args[0], f"passed straight to {node.func.attr}()")
        self.generic_visit(node)

    def visit_Assign(self, node: ast.Assign) -> None:
        for target in node.targets:
            if isinstance(target, ast.Name) and names_sql(target.id):
                self.inspect(node.value, f"stored in {target.id}")
        self.generic_visit(node)

    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
        if isinstance(node.target, ast.Name) and names_sql(node.target.id) and node.value:
            self.inspect(node.value, f"stored in {node.target.id}")
        self.generic_visit(node)


def scan(directory: Path) -> list[str]:
    findings: list[str] = []
    for path in sorted(directory.rglob("*.py")):
        if path.name in EXEMPT:
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        scanner = Scanner(path)
        scanner.visit(tree)
        findings.extend(scanner.findings)
    return findings


def main() -> int:
    directory = Path(sys.argv[1] if len(sys.argv) > 1 else Path(__file__).parent)
    checked = [p.name for p in sorted(directory.rglob("*.py")) if p.name not in EXEMPT]
    findings = scan(directory)

    print(f"Scanning {len(checked)} Python file(s) under {directory.name}/ for assembled SQL.")
    for name in checked:
        print(f"    {name}")
    print(f"Exempt by name (they build broken SQL on purpose): {', '.join(sorted(EXEMPT))}")
    print()
    if findings:
        for finding in findings:
            print(f"  FAIL: {finding}")
        print()
        print(f"{len(findings)} SQL string(s) built from parts. Bind the values instead.")
        return 1
    print("  ok: every SQL statement is a literal; every value is bound.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""The application layer. Look at what it does not import.

There is no `sqlite3` here, no SQL string, and no connection object built by
hand. This module asks a repository for domain objects and prints them. It
would work unchanged against a repository backed by a different database, a
file, or a fake built for a test — which is the argument Day 74 made about
boundaries, now cashed in.

Run it:  python3 report.py
"""

from __future__ import annotations

import shutil
import sys
import tempfile
from contextlib import closing
from pathlib import Path

import seed
from db import BookRepository, LoanRepository
from domain import Book, BookNotFound, DuplicateTitle, LibraryError


def shelf_report(books: BookRepository) -> list[str]:
    """Pure formatting over domain objects. Nothing here can fail on I/O."""
    lines = ["Shelf, by year:"]
    for book in sorted(books.all_sorted("year"), key=lambda b: b.year):
        status = "on the shelf" if book.copies else "all copies out"
        lines.append(f"  {book.year}  {book.title:<34} {book.copies} × ({status})")
    return lines


def overdue_report(loans: LoanRepository, as_of: str) -> list[str]:
    lines = [f"Overdue as of {as_of}:"]
    rows = loans.overdue(as_of)
    if not rows:
        lines.append("  nothing is overdue")
    for row in rows:
        lines.append(
            f"  {row['borrower']:<14} {row['book']:<32} due {row['due']}"
            f"  ({row['days_late']} days late)"
        )
    return lines


def add_with_domain_errors(books: BookRepository, book: Book) -> str:
    """Show the boundary translating storage errors into domain errors.

    The caller never sees an sqlite3 exception. It sees DuplicateTitle,
    which is a sentence about the library rather than about the engine.
    """
    try:
        stored = books.add(book)
    except DuplicateTitle as error:
        return f"  refused: {error}"
    return f"  stored as book_id {stored.book_id}: {stored.label}"


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-report-"))
    try:
        with closing(seed.build(sandbox / "library.db")) as connection:
            books = BookRepository(connection)
            loans = LoanRepository(connection)

            for line in shelf_report(books):
                print(line)
            print()
            for line in overdue_report(loans, seed.AS_OF):
                print(line)

            print()
            print("Adding books through the repository:")
            print(add_with_domain_errors(
                books, Book(title="Compilers", author="Alfred Aho", year=1986, copies=1)))
            print(add_with_domain_errors(
                books, Book(title="Compilers", author="Alfred Aho", year=1986, copies=1)))

            print()
            print("Looking one up that is not there:")
            try:
                books.get(4242)
            except LibraryError as error:
                print(f"  {type(error).__name__}: {error}")

            print()
            print("Sorting by a key chosen at runtime:")
            for key in ("author", "nonsense; DROP TABLE books"):
                try:
                    first = books.all_sorted(key)[0]
                    print(f"  sort_key={key!r:<32} first row: {first.title}")
                except ValueError as error:
                    print(f"  sort_key={key!r:<32} {error}")

            print()
            print("Streaming every book without building a list:")
            titles = [book.title for book in books.stream_all()]
            print(f"  {len(titles)} titles, one row held at a time")
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Fixed sample data, and one function that builds a database from it.

Every date here is a literal. Nothing calls `date.today()`, because a
fixture that moves makes today's captured output stop matching tomorrow's
run for no reason anybody can debug.
"""

from __future__ import annotations

from pathlib import Path

from db import BookRepository, LoanRepository, apply_schema, connect, transaction
from domain import Book, Member

BOOKS = [
    Book(title="The Art of Computer Programming", author="Donald Knuth", year=1968, copies=2),
    Book(title="A Relational Model of Data", author="Edgar Codd", year=1970, copies=3),
    Book(title="The Mythical Man-Month", author="Fred Brooks", year=1975, copies=1),
    Book(title="A Discipline of Programming", author="Edsger Dijkstra", year=1976, copies=2),
    Book(title="Structure and Interpretation", author="Harold Abelson", year=1985, copies=4),
    Book(title="The Practice of Programming", author="Brian Kernighan", year=1999, copies=2),
    Book(title="Programming Pearls", author="Jon Bentley", year=1986, copies=1),
]

MEMBERS = [
    Member(name="Ada Lovelace", email="ada@example.invalid"),
    Member(name="Grace Hopper", email="grace@example.invalid"),
    Member(name="Alan Turing", email="alan@example.invalid"),
]

# (book_id, member_id, borrowed_on, due_on)
LOANS = [
    (3, 1, "2026-06-08", "2026-06-22"),
    (4, 2, "2026-07-12", "2026-07-26"),
    (5, 1, "2026-07-27", "2026-08-10"),
    (1, 3, "2026-08-14", "2026-08-28"),
]

AS_OF = "2026-08-16"


def build(path: str | Path):
    """Create and populate a database, and hand back an open connection.

    The whole seed happens inside one transaction: either every row lands or
    none does. A half-seeded fixture is worse than no fixture.
    """
    connection = connect(path)
    apply_schema(connection)
    books = BookRepository(connection)
    loans = LoanRepository(connection)
    with transaction(connection):
        books.add_many(BOOKS)
        for member in MEMBERS:
            loans.add_member(member)
        for book_id, member_id, borrowed_on, due_on in LOANS:
            loans.borrow(book_id, member_id, borrowed_on, due_on)
    return connection

"""The data layer's own test suite, against a real database in a temp file.

Two decisions are worth arguing with, because both are deliberate.

**A real database file, not a mock.** Day 74 said that fakes beat mocks and
that the best move is usually to relocate the boundary rather than patch
across it. That is exactly what a repository does. So these tests do not
mock `sqlite3` — mocking it would test that the code calls the functions the
author expected, which is a tautology. They open a real SQLite database in a
temporary directory, which costs a few milliseconds and tests the thing that
can actually be wrong: the SQL.

**A file, not `:memory:`.** An in-memory database is faster and cannot test
anything about files, paths or durability. Using a file in `tempfile` keeps
the tests honest and still leaves nothing behind, because `tearDown` removes
the directory.

Run it:  python3 test_repository.py          (verbose: -v)
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path

import seed
from db import BookRepository, LoanRepository, apply_schema, connect, transaction
from domain import Book, BookNotFound, DuplicateTitle, InvalidBook, Member


class RepositoryTestCase(unittest.TestCase):
    """A fresh, seeded database per test. Tests never share state."""

    def setUp(self) -> None:
        self.sandbox = Path(tempfile.mkdtemp(prefix="day090-tests-"))
        self.path = self.sandbox / "library.db"
        self.connection = seed.build(self.path)
        self.books = BookRepository(self.connection)
        self.loans = LoanRepository(self.connection)

    def tearDown(self) -> None:
        self.connection.close()
        shutil.rmtree(self.sandbox, ignore_errors=True)


class TestConnectionFactory(RepositoryTestCase):
    def test_database_is_an_ordinary_file_where_we_asked(self) -> None:
        self.assertTrue(self.path.is_file())
        self.assertGreater(self.path.stat().st_size, 0)

    def test_foreign_keys_are_on_for_this_connection(self) -> None:
        self.assertEqual(self.connection.execute("PRAGMA foreign_keys").fetchone()[0], 1)

    def test_foreign_keys_are_off_on_a_connection_that_did_not_ask(self) -> None:
        """The pragma is per connection, not a property of the file."""
        raw = sqlite3.connect(self.path)
        try:
            self.assertEqual(raw.execute("PRAGMA foreign_keys").fetchone()[0], 0)
            # And the consequence: the same bad write is accepted.
            raw.execute(
                "INSERT INTO loans (book_id, member_id, borrowed_on, due_on)"
                " VALUES (?, ?, ?, ?)",
                (1, 999, "2026-08-01", "2026-08-15"),
            )
            raw.rollback()
        finally:
            raw.close()

    def test_the_configured_connection_refuses_that_same_write(self) -> None:
        with self.assertRaises(sqlite3.IntegrityError):
            self.connection.execute(
                "INSERT INTO loans (book_id, member_id, borrowed_on, due_on)"
                " VALUES (?, ?, ?, ?)",
                (1, 999, "2026-08-01", "2026-08-15"),
            )

    def test_rows_are_addressable_by_name(self) -> None:
        row = self.connection.execute("SELECT title, year FROM books LIMIT 1").fetchone()
        self.assertIsInstance(row, sqlite3.Row)
        self.assertEqual(row["title"], row[0])


class TestMapping(RepositoryTestCase):
    def test_rows_become_domain_objects(self) -> None:
        book = self.books.get(3)
        self.assertIsInstance(book, Book)
        self.assertEqual(book.title, "The Mythical Man-Month")
        self.assertEqual(book.year, 1975)
        self.assertEqual(book.label, "The Mythical Man-Month (Fred Brooks, 1975)")

    def test_missing_row_raises_a_domain_error_not_a_storage_one(self) -> None:
        with self.assertRaises(BookNotFound):
            self.books.get(4242)

    def test_add_returns_the_book_with_the_id_the_database_assigned(self) -> None:
        stored = self.books.add(Book(title="Compilers", author="Alfred Aho", year=1986, copies=1))
        self.assertIsNotNone(stored.book_id)
        self.assertEqual(self.books.get(stored.book_id).title, "Compilers")

    def test_duplicate_title_is_translated_at_the_boundary(self) -> None:
        with self.assertRaises(DuplicateTitle):
            self.books.add(
                Book(title="Programming Pearls", author="Jon Bentley", year=1986, copies=1)
            )

    def test_the_domain_refuses_a_bad_object_before_any_sql_runs(self) -> None:
        with self.assertRaises(InvalidBook):
            Book(title="Impossible", author="Nobody", year=1975, copies=-4)


class TestParameterBinding(RepositoryTestCase):
    HOSTILE = "Fred Brooks' OR '1'='1"
    DESTRUCTIVE = "Fred Brooks'; DROP TABLE books; --"

    def test_a_crafted_value_is_treated_as_ordinary_text(self) -> None:
        self.assertEqual(self.books.find_by_author(self.HOSTILE), [])
        self.assertEqual(self.books.find_by_author(self.DESTRUCTIVE), [])

    def test_the_tables_survive_it(self) -> None:
        self.books.find_by_author(self.DESTRUCTIVE)
        self.assertEqual(self.books.count(), len(seed.BOOKS))

    def test_an_ordinary_value_still_works(self) -> None:
        found = self.books.find_by_author("Fred Brooks")
        self.assertEqual([book.title for book in found], ["The Mythical Man-Month"])

    def test_named_placeholders(self) -> None:
        found = self.books.published_between(1968, 1976)
        self.assertEqual([book.year for book in found], [1968, 1970, 1975, 1976])

    def test_mixing_named_placeholders_with_a_sequence_is_an_error(self) -> None:
        with self.assertRaises(sqlite3.ProgrammingError):
            self.connection.execute("SELECT * FROM books WHERE year = :year", (1975,))

    def test_a_placeholder_cannot_stand_in_for_a_column_name(self) -> None:
        """ORDER BY ? binds a VALUE, so every row sorts by the same constant."""
        rows = self.connection.execute(
            "SELECT title FROM books ORDER BY ?", ("year",)
        ).fetchall()
        by_year = self.connection.execute("SELECT title FROM books ORDER BY year").fetchall()
        self.assertNotEqual([r["title"] for r in rows], [r["title"] for r in by_year])

    def test_sorting_uses_an_allow_list(self) -> None:
        self.assertEqual(self.books.all_sorted("year")[0].year, 1968)
        with self.assertRaises(ValueError):
            self.books.all_sorted("year; DROP TABLE books")


class TestTransactions(RepositoryTestCase):
    def test_a_failure_mid_transaction_leaves_the_database_unchanged(self) -> None:
        copies_before = self.books.get(1).copies
        loans_before = self.loans.open_count()

        with self.assertRaises(sqlite3.IntegrityError):
            with transaction(self.connection):
                self.loans.borrow(1, 1, "2026-08-01", "2026-08-15")
                self.loans.borrow(2, 999, "2026-08-01", "2026-08-15")  # no member 999

        self.assertEqual(self.books.get(1).copies, copies_before)
        self.assertEqual(self.loans.open_count(), loans_before)

    def test_a_python_error_rolls_back_just_as_a_sql_one_does(self) -> None:
        loans_before = self.loans.open_count()
        with self.assertRaises(ZeroDivisionError):
            with transaction(self.connection):
                self.loans.borrow(1, 1, "2026-08-01", "2026-08-15")
                _ = 1 / 0
        self.assertEqual(self.loans.open_count(), loans_before)

    def test_a_successful_transaction_is_visible_to_another_connection(self) -> None:
        with transaction(self.connection):
            self.loans.borrow(1, 1, "2026-08-01", "2026-08-15")
        other = connect(self.path)
        try:
            self.assertEqual(LoanRepository(other).open_count(), self.loans.open_count())
        finally:
            other.close()

    def test_the_transaction_closes_itself_either_way(self) -> None:
        self.assertFalse(self.connection.in_transaction)
        with transaction(self.connection):
            self.assertTrue(self.connection.in_transaction)
        self.assertFalse(self.connection.in_transaction)

    def test_nesting_is_refused_rather_than_silently_wrong(self) -> None:
        with transaction(self.connection):
            with self.assertRaises(RuntimeError):
                with transaction(self.connection):
                    pass

    def test_with_connection_does_not_close_the_connection(self) -> None:
        """The misreading this lab exists partly to correct."""
        with self.connection:
            self.connection.execute("SELECT 1")
        self.connection.execute("SELECT 1").fetchone()  # would raise if it were closed


class TestBulkAndStreaming(RepositoryTestCase):
    def test_executemany_stores_every_row(self) -> None:
        extra = [
            Book(title=f"Volume {n}", author="Anon", year=1990, copies=1)
            for n in range(500)
        ]
        with transaction(self.connection):
            written = self.books.add_many(extra)
        self.assertEqual(written, 500)
        self.assertEqual(self.books.count(), len(seed.BOOKS) + 500)

    def test_executemany_is_atomic_inside_a_transaction(self) -> None:
        before = self.books.count()
        clashing = [
            Book(title="Fresh Title", author="Anon", year=1990, copies=1),
            Book(title="Programming Pearls", author="Anon", year=1990, copies=1),  # taken
        ]
        with self.assertRaises(sqlite3.IntegrityError):
            with transaction(self.connection):
                self.books.add_many(clashing)
        self.assertEqual(self.books.count(), before)

    def test_streaming_yields_the_same_books_as_fetching_them_all(self) -> None:
        streamed = [book.title for book in self.books.stream_all()]
        fetched = [book.title for book in self.books.all_sorted("title")]
        self.assertEqual(sorted(streamed), sorted(fetched))
        self.assertEqual(len(streamed), len(seed.BOOKS))


class TestReports(RepositoryTestCase):
    def test_overdue_is_computed_from_the_date_we_pass_in(self) -> None:
        rows = self.loans.overdue(seed.AS_OF)
        self.assertEqual([row["days_late"] for row in rows], [55, 21, 6])

    def test_a_different_as_of_date_gives_a_different_answer(self) -> None:
        self.assertEqual(self.loans.overdue("2026-06-01"), [])

    def test_the_report_never_calls_todays_date(self) -> None:
        """A fixture that moves is a test that fails on a Tuesday."""
        source = Path(__file__).with_name("db.py").read_text(encoding="utf-8")
        for forbidden in ("date.today", "datetime.now", "date('now')", "CURRENT_DATE"):
            self.assertNotIn(forbidden, source)


if __name__ == "__main__":
    unittest.main(verbosity=2 if "-v" in sys.argv else 1)

"""The transaction model, demonstrated on the interpreter you are running.

This is the part of `sqlite3` that has changed most across Python versions,
so nothing here is asserted from memory: every line prints what THIS
interpreter actually does, and the test suite checks the behaviour rather
than the version number.

Run it:  python3 transactions_demo.py
"""

from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
from contextlib import closing
from pathlib import Path

from db import BookRepository, LoanRepository, apply_schema, connect, transaction
from domain import Book, Member


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


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-transactions-"))
    try:
        return run(sandbox)
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)


def run(sandbox: Path) -> int:
    print(f"python {sys.version.split()[0]}   sqlite3.sqlite_version {sqlite3.sqlite_version}")

    rule("1. the module's implicit transaction handling, watched")
    path = sandbox / "implicit.db"
    with closing(sqlite3.connect(path)) as connection:
        print(f"    default isolation_level: {connection.isolation_level!r}"
              "   (empty string = the module manages transactions for you)")
        print(f"    connection.autocommit:   {connection.autocommit}"
              f"   (== sqlite3.LEGACY_TRANSACTION_CONTROL, which is"
              f" {sqlite3.LEGACY_TRANSACTION_CONTROL})")
        print(f"    in_transaction on a fresh connection: {connection.in_transaction}")

        connection.execute("CREATE TABLE t (n INTEGER) STRICT")
        print(f"    after CREATE TABLE (DDL):             {connection.in_transaction}"
              "   — no transaction was opened")
        connection.execute("INSERT INTO t (n) VALUES (1)")
        print(f"    after INSERT (DML):                   {connection.in_transaction}"
              "   — the module opened one for you")
        connection.rollback()
        remaining = connection.execute("SELECT count(*) AS n FROM t").fetchone()[0]
        print(f"    after rollback(), rows in t:          {remaining}")
        print("    That INSERT was never committed. A program that forgets to")
        print("    commit and then exits loses the write, and nothing warns it.")

    rule("2. what `with connection:` does — and the two things it does not")
    path = sandbox / "withblock.db"
    connection = sqlite3.connect(path)
    connection.execute("CREATE TABLE t (n INTEGER) STRICT")
    with connection:
        connection.execute("INSERT INTO t (n) VALUES (1)")
    print(f"    after a successful with-block, in_transaction: {connection.in_transaction}"
          "   (committed)")

    try:
        with connection:
            connection.execute("INSERT INTO t (n) VALUES (2)")
            raise ValueError("something went wrong halfway through")
    except ValueError as error:
        print(f"    the block raised: {error}")
    kept = [row[0] for row in connection.execute("SELECT n FROM t ORDER BY n")]
    print(f"    rows in t now: {kept}   — the second insert was rolled back")

    still_open = True
    try:
        connection.execute("SELECT 1").fetchone()
    except sqlite3.ProgrammingError:
        still_open = False
    print(f"    is the connection still usable after the with-block? {still_open}")
    print("    THIS IS THE MISREADING WORTH KILLING: `with connection:` commits or")
    print("    rolls back a TRANSACTION. It does not close the connection. For")
    print("    closing, use contextlib.closing(sqlite3.connect(...)) — or both,")
    print("    nested, which is the honest full form.")
    connection.close()

    rule("3. explicit control: isolation_level=None and a context manager you wrote")
    path = sandbox / "explicit.db"
    with closing(connect(path)) as connection:
        apply_schema(connection)
        print(f"    connect() sets isolation_level = {connection.isolation_level!r}"
              "   — nothing implicit remains")
        books = BookRepository(connection)
        loans = LoanRepository(connection)

        with transaction(connection):
            books.add_many(
                [
                    Book(title="The Mythical Man-Month", author="Fred Brooks", year=1975, copies=1),
                    Book(title="Programming Pearls", author="Jon Bentley", year=1986, copies=2),
                ]
            )
            loans.add_member(Member(name="Ada Lovelace", email="ada@example.invalid"))
        print(f"    seeded inside one transaction: {books.count()} books")

        rule("4. all-or-nothing across two writes, proved by breaking the second")
        before_copies = books.get(1).copies
        before_loans = loans.open_count()
        print(f"    before: book 1 has {before_copies} copy/copies, {before_loans} open loan(s)")
        try:
            with transaction(connection):
                loans.borrow(1, 1, "2026-08-01", "2026-08-15")
                # Member 999 does not exist. PRAGMA foreign_keys is ON, so
                # this write is refused — after the loan row and the
                # decremented copy count are already in the transaction.
                loans.borrow(2, 999, "2026-08-01", "2026-08-15")
        except sqlite3.IntegrityError as error:
            print(f"    the transaction raised: {type(error).__name__}: {error}")
        after_copies = books.get(1).copies
        after_loans = loans.open_count()
        print(f"    after:  book 1 has {after_copies} copy/copies, {after_loans} open loan(s)")
        print("    Both halves of the FIRST borrow were undone as well. That is")
        print("    atomicity: the group either lands or it does not.")

        rule("5. the pragma trap: PRAGMA foreign_keys is a no-op inside a transaction")
        connection.execute("PRAGMA foreign_keys = OFF")
        print(f"    outside a transaction, set OFF -> "
              f"{connection.execute('PRAGMA foreign_keys').fetchone()[0]}")
        connection.execute("BEGIN")
        connection.execute("PRAGMA foreign_keys = ON")
        inside = connection.execute("PRAGMA foreign_keys").fetchone()[0]
        print(f"    inside a transaction, set ON   -> {inside}"
              "   (silently ignored — no error, no warning)")
        connection.commit()
        connection.execute("PRAGMA foreign_keys = ON")
        outside = connection.execute("PRAGMA foreign_keys").fetchone()[0]
        print(f"    outside again, set ON          -> {outside}")
        print("    This is why connect() runs the pragma the moment the")
        print("    connection is opened, before anything can begin a transaction.")

    rule("6. connection.autocommit — the newer, explicit control")
    path = sandbox / "autocommit.db"
    with closing(sqlite3.connect(path)) as writer, closing(sqlite3.connect(path)) as reader:
        writer.execute("CREATE TABLE t (n INTEGER) STRICT")
        writer.commit()

        writer.autocommit = True
        writer.execute("INSERT INTO t (n) VALUES (1)")
        seen = reader.execute("SELECT count(*) AS n FROM t").fetchone()[0]
        print(f"    autocommit = True:  in_transaction {writer.in_transaction},"
              f" another connection already sees {seen} row(s) — committed immediately")

        writer.autocommit = False
        writer.execute("INSERT INTO t (n) VALUES (2)")
        seen = reader.execute("SELECT count(*) AS n FROM t").fetchone()[0]
        print(f"    autocommit = False: in_transaction {writer.in_transaction},"
              f" the other connection still sees {seen} row(s) — uncommitted")
        writer.rollback()
        seen = reader.execute("SELECT count(*) AS n FROM t").fetchone()[0]
        print(f"    after rollback():   the other connection sees {seen} row(s)")
        print("    autocommit=False opens a transaction and keeps one open, so a")
        print("    long-lived connection holds a read lock until you commit.")
        writer.autocommit = sqlite3.LEGACY_TRANSACTION_CONTROL
        print(f"    set back to LEGACY_TRANSACTION_CONTROL"
              f" ({sqlite3.LEGACY_TRANSACTION_CONTROL}): isolation_level is honoured again")

    print()
    print("sandbox removed.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

lesson_id: D090
day: 90
kind: guided-build
languages: [python, sql, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-090-sqlite-from-python
  - 'python3 -c "import sqlite3, sys; print(sys.version.split()[0], sqlite3.sqlite_version)"'
run_commands:
  - python3 examples/injection_demo.py
  - python3 examples/transactions_demo.py
  - python3 examples/cursors_demo.py
  - python3 examples/errors_demo.py
  - 'python3 examples/bulk_insert.py 2000  # the default of 20000 takes about 15 seconds'
  - python3 examples/report.py
  - python3 examples/test_repository.py -v
  - python3 examples/no_sql_strings.py examples
  - 'cd starter && python3 smoke.py  # exits 1 until all nine exercises are written'
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - 'find "${TMPDIR:-/tmp}" -maxdepth 1 -name ''day090-*''  # look first; scripts clean up after themselves'
  - find . -type d -name __pycache__ -prune -exec rm -rf -- {} +
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: false
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.1 (Apple Silicon, arm64), Python 3.14.0, SQLite 3.53.3 as linked into Python, bash 3.2.57 — bash tests/run_tests.sh -> 64 checks, 0 failure(s), exit 0'

# What this lab needs, and where it comes from

Nothing to install. Every import in every file is standard library.

## The one thing you need

| Thing | Where it comes from | Cost | How to check |
| --- | --- | --- | --- |
| Python 3.12 or newer, with `sqlite3` | Part of Python since 2.5. You already have it | Free; part of Python | `python3 -c "import sqlite3; print(sqlite3.sqlite_version)"` |
| `bash`, for the test harness | Preinstalled on macOS and Linux; WSL or Git Bash on Windows | Free | `bash --version` |

Captures were taken on **Python 3.14.0** with the module linked against
**SQLite 3.53.3**.

## Why 3.12 rather than 3.11

Two things in this lab are newer than 3.11 and both are checked by the test
suite rather than assumed:

- **`Connection.autocommit`** arrived in Python 3.12. `transactions_demo.py`
  demonstrates it directly, and section 1 of the harness asserts the
  attribute exists. On 3.11 that check fails honestly rather than the demo
  crashing halfway through.
- **The default date and timestamp adapters are deprecated as of Python
  3.12.** The lesson reports the exact deprecation message this interpreter
  produces. Nothing in the lab relies on those adapters — every date is
  stored as an ISO-8601 `TEXT` string, which is what the deprecation notes
  recommend you do instead.

`STRICT` tables need **SQLite 3.37.0 or newer**, which the harness also
checks. That is the library linked into Python, not a separate install.

## Two SQLite version numbers on one machine

If you also have the `sqlite3` command-line shell, `sqlite3 --version` may
print a different number from `python3 -c "import sqlite3;
print(sqlite3.sqlite_version)"`. That is normal: two programs, each linking
its own copy of the library, both reading and writing the same file format.
This lab needs only the Python one. Day 85's lab covers the shell.

## What the lab deliberately does not use

- **No ORM.** SQLAlchemy Core and the SQLAlchemy ORM are excellent, and Day
  93 covers them properly. Today you write the layer they would replace, so
  that when you do reach for one you know what it is doing for you.
- **No `pytest`.** `unittest` is in the standard library and does everything
  this suite needs. `tests/run_tests.sh` is a bash harness that runs it.
- **No `aiosqlite`, no `pandas`.** Both appear in the lesson's Alternatives
  section, described rather than demonstrated. Neither is installed for this
  lab, and no output from either is claimed anywhere.
- **No network, asserted mechanically.** The harness fails if any file under
  `examples/` or `starter/` contains a URL or an IP address.
- **No `sudo`, no server, no port, no credential.**

# Day 090 — A Real Data Layer
#
# This file is deliberately empty of packages.
#
# Everything this lab needs is in the Python standard library: sqlite3,
# contextlib, dataclasses, tempfile, unittest, ast and tracemalloc. There is
# no pip install, no virtual environment, no network and no account.
#
# It exists so that `pip install -r requirements/requirements.txt` succeeds
# and does nothing, and so that the absence of dependencies is a written-down
# decision rather than an omission somebody has to guess about.
#
# The lesson's Alternatives section covers SQLAlchemy Core, the SQLAlchemy
# ORM, aiosqlite and pandas honestly — including the fact that none of them
# is installed here, so no output from any of them is claimed anywhere.
#
# See requirements/README.md for what the lab uses and where it comes from.

"""YOUR data layer. Nine numbered exercises.

`domain.py` and `seed.py` beside this file are finished and need no changes.
Everything you write goes here.

How to work:

    python3 smoke.py          # names the next unfinished exercise, exits 1
    python3 smoke.py          # ...until it exits 0
    python3 test_repository.py    # then the real suite, from examples/

Every exercise says exactly which call to use. None of them needs more than
about five lines. The one rule that applies to all nine: **no value ever
goes into a SQL string.** Statements are literals; values are bound.
"""

from __future__ import annotations

import sqlite3
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from pathlib import Path

from domain import Book, BookNotFound, DuplicateTitle, Loan, Member

# Given: the schema. Read it before you start — every exercise below is
# constrained by something written here.
SCHEMA = """
CREATE TABLE IF NOT EXISTS books (
    book_id  INTEGER PRIMARY KEY,
    title    TEXT    NOT NULL UNIQUE,
    author   TEXT    NOT NULL,
    year     INTEGER NOT NULL CHECK (year BETWEEN 1400 AND 2100),
    copies   INTEGER NOT NULL CHECK (copies >= 0)
) STRICT;

CREATE TABLE IF NOT EXISTS members (
    member_id INTEGER PRIMARY KEY,
    name      TEXT NOT NULL,
    email     TEXT NOT NULL UNIQUE
) STRICT;

CREATE TABLE IF NOT EXISTS loans (
    loan_id     INTEGER PRIMARY KEY,
    book_id     INTEGER NOT NULL REFERENCES books(book_id),
    member_id   INTEGER NOT NULL REFERENCES members(member_id),
    borrowed_on TEXT NOT NULL,
    due_on      TEXT NOT NULL,
    returned_on TEXT,
    CHECK (due_on >= borrowed_on)
) STRICT;

CREATE INDEX IF NOT EXISTS loans_open ON loans(due_on) WHERE returned_on IS NULL;
"""


# ===========================================================================
# EXERCISE 1 — the connection factory
# ===========================================================================
def connect(path: str | Path) -> sqlite3.Connection:
    """Open a connection and configure it.

    Do four things, in this order:

      1. `sqlite3.connect(str(path), isolation_level=None, timeout=5.0)`
         — isolation_level=None turns off the module's implicit transaction
         handling, so `transaction()` below is the only thing that opens one.
      2. set `connection.row_factory = sqlite3.Row` so rows are addressable
         by column name.
      3. `connection.execute("PRAGMA foreign_keys = ON")` — foreign keys are
         OFF by default, per connection. Do it here, before anything can
         start a transaction: the pragma is a silent no-op inside one.
      4. return the connection.

    Checked by: smoke.py step 1, and
    TestConnectionFactory in examples/test_repository.py.
    """
    raise NotImplementedError("exercise 1: connect()")


def dict_factory(cursor: sqlite3.Cursor, row: tuple) -> dict:
    """Given, as a worked model of what a row factory actually is.

    A row factory is any callable taking (cursor, row) and returning
    whatever you want a row to be. `cursor.description` is a 7-tuple per
    column of which sqlite3 fills in only the first item, the name.
    """
    return {column[0]: value for column, value in zip(cursor.description, row)}


def apply_schema(connection: sqlite3.Connection) -> None:
    """Given. Note that `executescript` takes no parameters at all, and
    issues an implicit COMMIT before it runs — so it can never be nested
    inside a transaction."""
    connection.executescript(SCHEMA)


# ===========================================================================
# EXERCISE 2 — the transaction context manager
# ===========================================================================
@contextmanager
def transaction(connection: sqlite3.Connection) -> Iterator[sqlite3.Connection]:
    """Begin, then commit on success and roll back on anything at all.

      1. if `connection.in_transaction` is already True, raise RuntimeError
         with a message saying SQLite has no nested transactions.
      2. `connection.execute("BEGIN")`.
      3. `yield connection` inside a try.
      4. `except BaseException:` roll back and re-raise. BaseException, not
         Exception: a KeyboardInterrupt halfway through a two-statement
         change must roll back too.
      5. `else:` commit.

    Checked by: smoke.py step 2, TestTransactions.
    """
    raise NotImplementedError("exercise 2: transaction()")


# ===========================================================================
# EXERCISE 3 — row-to-object mapping
# ===========================================================================
def row_to_book(row: sqlite3.Row) -> Book:
    """Build one Book from one row, addressing columns BY NAME.

    Use `row["book_id"]`, not `row[0]`. Positional access is a bug waiting
    for somebody to add a column to the SELECT list.

    Checked by: smoke.py step 3, TestMapping.
    """
    raise NotImplementedError("exercise 3: row_to_book()")


def row_to_member(row: sqlite3.Row) -> Member:
    """Given, as the model for exercise 3."""
    return Member(member_id=row["member_id"], name=row["name"], email=row["email"])


def row_to_loan(row: sqlite3.Row) -> Loan:
    """Given."""
    return Loan(
        loan_id=row["loan_id"],
        book_id=row["book_id"],
        member_id=row["member_id"],
        borrowed_on=row["borrowed_on"],
        due_on=row["due_on"],
        returned_on=row["returned_on"],
    )


# Given: sorting has to vary at runtime, and a placeholder cannot stand in
# for an identifier — `ORDER BY ?` binds a VALUE. The safe form is an
# allow-list of whole statements you wrote.
SORTED_QUERIES = {
    "title": "SELECT book_id, title, author, year, copies FROM books ORDER BY title",
    "author": "SELECT book_id, title, author, year, copies FROM books ORDER BY author",
    "year": "SELECT book_id, title, author, year, copies FROM books ORDER BY year",
}


class BookRepository:
    """Every SQL statement about books lives in this class, and nothing else does."""

    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection

    # =======================================================================
    # EXERCISE 4 — a read with one bound parameter
    # =======================================================================
    def get(self, book_id: int) -> Book:
        """SELECT one book by id.

        `connection.execute("SELECT ... WHERE book_id = ?", (book_id,))`,
        then `.fetchone()`. If the result is None, raise BookNotFound —
        never return None, because a caller will forget to check it.
        Otherwise return `row_to_book(row)`.

        Note the trailing comma in `(book_id,)`. Without it that is not a
        tuple, and you get a ProgrammingError about bindings.

        Checked by: smoke.py step 4, TestMapping.
        """
        raise NotImplementedError("exercise 4: BookRepository.get()")

    # =======================================================================
    # EXERCISE 5 — the query the attacker aims at
    # =======================================================================
    def find_by_author(self, author: str) -> list[Book]:
        """SELECT every book by one author, ORDER BY year.

        `author` comes from outside the program. Bind it. Return a list of
        Book objects built by iterating the cursor:

            cursor = self._connection.execute("SELECT ... WHERE author = ?"
                                              " ORDER BY year", (author,))
            return [row_to_book(row) for row in cursor]

        Two adjacent string literals like that are joined by Python at
        compile time — no runtime value can enter, so it is the right way to
        wrap a long statement. An f-string is not.

        Checked by: smoke.py step 5, TestParameterBinding.
        """
        raise NotImplementedError("exercise 5: BookRepository.find_by_author()")

    # =======================================================================
    # EXERCISE 6 — named placeholders
    # =======================================================================
    def published_between(self, first: int, last: int) -> list[Book]:
        """SELECT books with year BETWEEN :first AND :last, ORDER BY year.

        Named style takes a MAPPING: `{"first": first, "last": last}`.
        Passing a tuple to named placeholders raises ProgrammingError —
        try it once on purpose so you recognise the message.

        Checked by: smoke.py step 6, TestParameterBinding.
        """
        raise NotImplementedError("exercise 6: BookRepository.published_between()")

    def all_sorted(self, sort_key: str = "title") -> list[Book]:
        """Given, as the model for handling an identifier chosen at runtime."""
        try:
            statement = SORTED_QUERIES[sort_key]
        except KeyError:
            raise ValueError(
                f"cannot sort by {sort_key!r}; choose from {sorted(SORTED_QUERIES)}"
            ) from None
        return [row_to_book(row) for row in self._connection.execute(statement)]

    # =======================================================================
    # EXERCISE 7 — streaming instead of fetching everything
    # =======================================================================
    def stream_all(self) -> Iterator[Book]:
        """Yield every book, one row at a time, without building a list.

        Execute the SELECT, then `for row in cursor: yield row_to_book(row)`.
        Do NOT call fetchall(). On a large table fetchall builds the whole
        result in memory; iterating holds one row.

        Checked by: smoke.py step 7, TestBulkAndStreaming.
        """
        raise NotImplementedError("exercise 7: BookRepository.stream_all()")

    def count(self) -> int:
        """Given."""
        return self._connection.execute("SELECT count(*) AS n FROM books").fetchone()["n"]

    # =======================================================================
    # EXERCISE 8 — a write, its assigned id, and the error translated
    # =======================================================================
    def add(self, book: Book) -> Book:
        """INSERT one book; return it with the id the database assigned.

          * `cursor = self._connection.execute("INSERT INTO books (title,
            author, year, copies) VALUES (?, ?, ?, ?)", (book.title,
            book.author, book.year, book.copies))`
          * the new id is `cursor.lastrowid` — which is why keeping the
            cursor that execute() returned is worth doing.
          * wrap it in `try / except sqlite3.IntegrityError as error:` and,
            when `"books.title" in str(error)`, raise
            `DuplicateTitle(...) from error`. Anything else, re-raise.
            That translation is the boundary doing its job: nothing above
            this line should have to import sqlite3 to learn that a title
            was taken.
          * return a NEW Book with book_id filled in. Book is frozen, so you
            build one rather than assigning to a field.

        Checked by: smoke.py step 8, TestMapping.
        """
        raise NotImplementedError("exercise 8: BookRepository.add()")

    # =======================================================================
    # EXERCISE 9 — the bulk write
    # =======================================================================
    def add_many(self, books: Sequence[Book]) -> int:
        """INSERT many books with ONE prepared statement.

        `cursor = self._connection.executemany(statement, sequence_of_tuples)`
        where the sequence is `[(b.title, b.author, b.year, b.copies) for b
        in books]`. Return `cursor.rowcount`.

        Call it inside `transaction()`; `examples/bulk_insert.py` measures
        why that matters more than executemany itself does.

        Checked by: smoke.py step 9, TestBulkAndStreaming.
        """
        raise NotImplementedError("exercise 9: BookRepository.add_many()")

    def set_copies(self, book_id: int, copies: int) -> None:
        """Given. Note `cursor.rowcount` here: for UPDATE and DELETE it is
        the number of rows changed, which is how you tell "updated nothing"
        from "updated something"."""
        cursor = self._connection.execute(
            "UPDATE books SET copies = ? WHERE book_id = ?", (copies, book_id)
        )
        if cursor.rowcount == 0:
            raise BookNotFound(f"no book with id {book_id}")

    def delete(self, book_id: int) -> None:
        """Given."""
        cursor = self._connection.execute("DELETE FROM books WHERE book_id = ?", (book_id,))
        if cursor.rowcount == 0:
            raise BookNotFound(f"no book with id {book_id}")


class LoanRepository:
    """Given in full. Read `borrow` — it is the reason transactions exist."""

    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection

    def add_member(self, member: Member) -> Member:
        cursor = self._connection.execute(
            "INSERT INTO members (name, email) VALUES (?, ?)", (member.name, member.email)
        )
        return Member(member_id=cursor.lastrowid, name=member.name, email=member.email)

    def borrow(self, book_id: int, member_id: int, borrowed_on: str, due_on: str) -> Loan:
        cursor = self._connection.execute(
            "INSERT INTO loans (book_id, member_id, borrowed_on, due_on) VALUES (?, ?, ?, ?)",
            (book_id, member_id, borrowed_on, due_on),
        )
        self._connection.execute(
            "UPDATE books SET copies = copies - 1 WHERE book_id = ?", (book_id,)
        )
        return Loan(
            loan_id=cursor.lastrowid,
            book_id=book_id,
            member_id=member_id,
            borrowed_on=borrowed_on,
            due_on=due_on,
        )

    def overdue(self, as_of: str) -> list[dict]:
        cursor = self._connection.execute(
            """
            SELECT   members.name AS borrower,
                     books.title  AS book,
                     loans.due_on AS due,
                     julianday(:as_of) - julianday(loans.due_on) AS days_late
            FROM     loans
            JOIN     books   ON books.book_id     = loans.book_id
            JOIN     members ON members.member_id = loans.member_id
            WHERE    loans.returned_on IS NULL
              AND    loans.due_on < :as_of
            ORDER BY days_late DESC
            """,
            {"as_of": as_of},
        )
        return [
            {
                "borrower": row["borrower"],
                "book": row["book"],
                "due": row["due"],
                "days_late": int(row["days_late"]),
            }
            for row in cursor
        ]

    def open_count(self) -> int:
        return self._connection.execute(
            "SELECT count(*) AS n FROM loans WHERE returned_on IS NULL"
        ).fetchone()["n"]

"""The domain objects — Day 70's model, unchanged and unaware of storage.

Read the imports at the top of this file. There is no `sqlite3` here, and
there never will be. That absence is the whole architectural claim of this
lab: the objects that carry your program's meaning do not know that a
database exists, which is exactly what makes them testable without one
(Day 74) and replaceable without rewriting them.

Everything that knows about SQL lives in `db.py`. Everything that knows
about the problem lives here.
"""

from __future__ import annotations

from dataclasses import dataclass


class LibraryError(Exception):
    """Base class for every error this domain raises.

    The outer layer catches this one class and can say something a human can
    act on, instead of leaking an `sqlite3.IntegrityError` from three layers
    down into a user interface.
    """


class InvalidBook(LibraryError):
    """A Book was asked to exist in a state the domain forbids."""


class BookNotFound(LibraryError):
    """A lookup by identity found nothing."""


class DuplicateTitle(LibraryError):
    """A write would have created a second book with an existing title."""


@dataclass(frozen=True)
class Book:
    """One book. Frozen, because a book's identity does not change.

    `book_id` is `None` for a book that has been built in memory but never
    stored. The repository fills it in when the database assigns one, by
    returning a new Book rather than mutating this one.
    """

    title: str
    author: str
    year: int
    copies: int
    book_id: int | None = None

    def __post_init__(self) -> None:
        if not self.title.strip():
            raise InvalidBook("a book must have a title")
        if not self.author.strip():
            raise InvalidBook("a book must have an author")
        if not isinstance(self.year, int) or not (1400 <= self.year <= 2100):
            raise InvalidBook(f"year out of range: {self.year!r}")
        if not isinstance(self.copies, int) or self.copies < 0:
            raise InvalidBook(f"copies must be a non-negative integer: {self.copies!r}")

    @property
    def label(self) -> str:
        """How a book prints in a report. Presentation, not persistence."""
        return f"{self.title} ({self.author}, {self.year})"


@dataclass(frozen=True)
class Member:
    """One library member."""

    name: str
    email: str
    member_id: int | None = None

    def __post_init__(self) -> None:
        if not self.name.strip():
            raise LibraryError("a member must have a name")
        if "@" not in self.email:
            raise LibraryError(f"not an address: {self.email!r}")


@dataclass(frozen=True)
class Loan:
    """One book, out with one member, due on one date.

    `returned_on` is None while the book is still out. That is the one place
    in this model where None means something specific rather than "missing".
    """

    book_id: int
    member_id: int
    borrowed_on: str
    due_on: str
    returned_on: str | None = None
    loan_id: int | None = None

    @property
    def is_open(self) -> bool:
        return self.returned_on is None

"""Fixed sample data, and one function that builds a database from it.

Every date here is a literal. Nothing calls `date.today()`, because a
fixture that moves makes today's captured output stop matching tomorrow's
run for no reason anybody can debug.
"""

from __future__ import annotations

from pathlib import Path

from db import BookRepository, LoanRepository, apply_schema, connect, transaction
from domain import Book, Member

BOOKS = [
    Book(title="The Art of Computer Programming", author="Donald Knuth", year=1968, copies=2),
    Book(title="A Relational Model of Data", author="Edgar Codd", year=1970, copies=3),
    Book(title="The Mythical Man-Month", author="Fred Brooks", year=1975, copies=1),
    Book(title="A Discipline of Programming", author="Edsger Dijkstra", year=1976, copies=2),
    Book(title="Structure and Interpretation", author="Harold Abelson", year=1985, copies=4),
    Book(title="The Practice of Programming", author="Brian Kernighan", year=1999, copies=2),
    Book(title="Programming Pearls", author="Jon Bentley", year=1986, copies=1),
]

MEMBERS = [
    Member(name="Ada Lovelace", email="ada@example.invalid"),
    Member(name="Grace Hopper", email="grace@example.invalid"),
    Member(name="Alan Turing", email="alan@example.invalid"),
]

# (book_id, member_id, borrowed_on, due_on)
LOANS = [
    (3, 1, "2026-06-08", "2026-06-22"),
    (4, 2, "2026-07-12", "2026-07-26"),
    (5, 1, "2026-07-27", "2026-08-10"),
    (1, 3, "2026-08-14", "2026-08-28"),
]

AS_OF = "2026-08-16"


def build(path: str | Path):
    """Create and populate a database, and hand back an open connection.

    The whole seed happens inside one transaction: either every row lands or
    none does. A half-seeded fixture is worse than no fixture.
    """
    connection = connect(path)
    apply_schema(connection)
    books = BookRepository(connection)
    loans = LoanRepository(connection)
    with transaction(connection):
        books.add_many(BOOKS)
        for member in MEMBERS:
            loans.add_member(member)
        for book_id, member_id, borrowed_on, due_on in LOANS:
            loans.borrow(book_id, member_id, borrowed_on, due_on)
    return connection

"""Runs your `db.py` and names the next unfinished exercise.

    python3 smoke.py

Exits 1 until all nine are written, on purpose: an unfinished lab should not
be able to look finished. Every database it builds lives in a temporary
directory that is removed on the way out.
"""

from __future__ import annotations

import ast
import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path

import db
from domain import Book, BookNotFound, DuplicateTitle, Member

STEPS: list[tuple[int, str, str]] = [
    (1, "connect()", "a configured connection: row factory, foreign keys on"),
    (2, "transaction()", "begin, commit on success, roll back on anything"),
    (3, "row_to_book()", "one row becomes one Book, addressed by column name"),
    (4, "BookRepository.get()", "one bound parameter, and BookNotFound when there is no row"),
    (5, "BookRepository.find_by_author()", "a hostile value bound, not interpolated"),
    (6, "BookRepository.published_between()", "named placeholders take a mapping"),
    (7, "BookRepository.stream_all()", "iterate the cursor, never fetchall"),
    (8, "BookRepository.add()", "lastrowid, and IntegrityError translated to DuplicateTitle"),
    (9, "BookRepository.add_many()", "executemany, one prepared statement"),
]

SAMPLE = [
    Book(title="The Mythical Man-Month", author="Fred Brooks", year=1975, copies=1),
    Book(title="A Discipline of Programming", author="Edsger Dijkstra", year=1976, copies=2),
    Book(title="Programming Pearls", author="Jon Bentley", year=1986, copies=1),
]


def calls_fetchall(function_name: str) -> bool:
    """Look at the code of one function in db.py, ignoring its docstring.

    Parsed with `ast` rather than searched as text, so a mention of
    "fetchall" in a comment or docstring does not count as a call to it.
    """
    tree = ast.parse(Path(db.__file__).read_text(encoding="utf-8"))
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == function_name:
            return any(
                isinstance(inner, ast.Call)
                and isinstance(inner.func, ast.Attribute)
                and inner.func.attr == "fetchall"
                for inner in ast.walk(node)
            )
    return False


def announce(number: int, detail: str = "") -> None:
    _, name, hint = STEPS[number - 1]
    print(f"\nEXERCISE {number} — {name}")
    print(f"  what it must do: {hint}")
    if detail:
        print(f"  what happened:   {detail}")
    print(f"  open db.py, find 'EXERCISE {number}', and write it. Then run this again.")


def main() -> int:
    sandbox = Path(tempfile.mkdtemp(prefix="day090-smoke-"))
    try:
        return run(sandbox)
    finally:
        shutil.rmtree(sandbox, ignore_errors=True)


def run(sandbox: Path) -> int:
    path = sandbox / "mine.db"
    done = 0

    # ---- 1 --------------------------------------------------------------
    try:
        connection = db.connect(path)
        assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1, (
            "connected, but PRAGMA foreign_keys is off on this connection"
        )
        db.apply_schema(connection)
        row = connection.execute("SELECT 1 AS one").fetchone()
        assert row["one"] == 1, "connected, but rows are not addressable by name"
    except NotImplementedError:
        announce(1)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(1, str(error))
        return summary(done)
    done = 1
    print("  ok: 1. connect() — foreign keys on, rows by name")

    books = db.BookRepository(connection)
    loans = db.LoanRepository(connection)

    # ---- 2 --------------------------------------------------------------
    try:
        with db.transaction(connection):
            connection.execute(
                "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                ("Temporary", "Nobody", 1999, 1),
            )
            raise RuntimeError("deliberate")
    except NotImplementedError:
        announce(2)
        return summary(done)
    except RuntimeError:
        pass
    left = connection.execute("SELECT count(*) FROM books").fetchone()[0]
    if left != 0:
        announce(2, f"the failed block left {left} row(s) behind — it did not roll back")
        return summary(done)
    done = 2
    print("  ok: 2. transaction() — a failure left nothing behind")

    # seed, using only the given pieces
    try:
        with db.transaction(connection):
            for book in SAMPLE:
                connection.execute(
                    "INSERT INTO books (title, author, year, copies) VALUES (?, ?, ?, ?)",
                    (book.title, book.author, book.year, book.copies),
                )
            loans.add_member(Member(name="Ada Lovelace", email="ada@example.invalid"))
    except sqlite3.Error as error:
        announce(2, f"seeding failed: {error}")
        return summary(done)

    # ---- 3 --------------------------------------------------------------
    row = connection.execute(
        "SELECT book_id, title, author, year, copies FROM books WHERE book_id = 1"
    ).fetchone()
    try:
        book = db.row_to_book(row)
        assert isinstance(book, Book), "row_to_book did not return a Book"
        assert book.title == "The Mythical Man-Month", f"got title {book.title!r}"
        assert book.book_id == 1, f"book_id should be 1, got {book.book_id!r}"
    except NotImplementedError:
        announce(3)
        return summary(done)
    except AssertionError as error:
        announce(3, str(error))
        return summary(done)
    done = 3
    print("  ok: 3. row_to_book() — a row became a domain object")

    # ---- 4 --------------------------------------------------------------
    try:
        assert books.get(2).title == "A Discipline of Programming"
        try:
            books.get(4242)
        except BookNotFound:
            pass
        else:
            raise AssertionError("get(4242) returned something instead of raising BookNotFound")
    except NotImplementedError:
        announce(4)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(4, str(error))
        return summary(done)
    done = 4
    print("  ok: 4. get() — found one, and raised BookNotFound for a missing id")

    # ---- 5 --------------------------------------------------------------
    hostile = "Fred Brooks' OR '1'='1"
    try:
        assert [b.title for b in books.find_by_author("Fred Brooks")] == [
            "The Mythical Man-Month"
        ], "the ordinary lookup did not find the one book by Fred Brooks"
        assert books.find_by_author(hostile) == [], (
            "the crafted value returned rows — the value is being interpolated, not bound"
        )
        assert books.count() == 3, "the crafted value changed the table"
    except NotImplementedError:
        announce(5)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(5, str(error))
        return summary(done)
    done = 5
    print("  ok: 5. find_by_author() — the crafted value was treated as data")

    # ---- 6 --------------------------------------------------------------
    try:
        years = [b.year for b in books.published_between(1975, 1980)]
        assert years == [1975, 1976], f"expected [1975, 1976], got {years}"
    except NotImplementedError:
        announce(6)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(6, str(error))
        return summary(done)
    done = 6
    print("  ok: 6. published_between() — named placeholders, in year order")

    # ---- 7 --------------------------------------------------------------
    try:
        streamed = list(books.stream_all())
        assert len(streamed) == 3, f"expected 3 books, got {len(streamed)}"
        assert all(isinstance(b, Book) for b in streamed), "stream_all yielded something else"
        assert not calls_fetchall("stream_all"), (
            "stream_all calls fetchall; iterate the cursor instead"
        )
    except NotImplementedError:
        announce(7)
        return summary(done)
    except (AssertionError, sqlite3.Error, IndexError) as error:
        announce(7, str(error))
        return summary(done)
    done = 7
    print("  ok: 7. stream_all() — three books, one row at a time")

    # ---- 8 --------------------------------------------------------------
    try:
        stored = books.add(Book(title="Compilers", author="Alfred Aho", year=1986, copies=1))
        assert stored.book_id is not None, "add() returned a Book with no book_id"
        assert books.get(stored.book_id).title == "Compilers"
        try:
            books.add(Book(title="Compilers", author="Alfred Aho", year=1986, copies=1))
        except DuplicateTitle:
            pass
        except sqlite3.IntegrityError:
            raise AssertionError(
                "the duplicate raised sqlite3.IntegrityError — translate it to DuplicateTitle"
            ) from None
        else:
            raise AssertionError("the duplicate title was accepted")
    except NotImplementedError:
        announce(8)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(8, str(error))
        return summary(done)
    done = 8
    print("  ok: 8. add() — id assigned, duplicate translated to DuplicateTitle")

    # ---- 9 --------------------------------------------------------------
    try:
        before = books.count()
        extra = [
            Book(title=f"Volume {n}", author="Anon", year=1990, copies=1) for n in range(50)
        ]
        with db.transaction(connection):
            written = books.add_many(extra)
        assert written == 50, f"add_many reported {written} rows, expected 50"
        assert books.count() == before + 50, "the rows are not in the table"
    except NotImplementedError:
        announce(9)
        return summary(done)
    except (AssertionError, sqlite3.Error) as error:
        announce(9, str(error))
        return summary(done)
    done = 9
    print("  ok: 9. add_many() — fifty rows, one prepared statement")

    connection.close()
    return summary(done)


def summary(done: int) -> int:
    print()
    print(f"{done} of {len(STEPS)} exercises finished.")
    if done == len(STEPS):
        print("All nine. Now run the real suite:  python3 ../examples/test_repository.py")
        print("(or, from the lab directory, bash tests/run_tests.sh)")
        return 0
    return 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# Tests for the Day 090 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# These checks ask whether the data layer actually has the properties it
# claims, rather than whether it runs:
#
#   * is a crafted value CODE when concatenated and DATA when bound — both
#     demonstrated on a throwaway database, both asserted?
#   * does a failure halfway through a transaction leave the database
#     exactly as it was, for a SQL error and for a Python one?
#   * is PRAGMA foreign_keys really per-connection, and really a no-op
#     inside a transaction?
#   * does `with connection:` leave the connection OPEN?
#   * is every SQL statement in the lab a literal, with every value bound?
#   * does executemany beat a loop, and does a transaction beat both?
#   * does the starter refuse to look finished before it is?
#
# Everything runs offline. No server, no port, no credential, no third-party
# package. Every database is created inside a directory made with mktemp -d
# and removed in a trap, so a completed run leaves nothing behind — and one
# of the checks asserts exactly that.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

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

yesno() { if [ "$1" -eq 0 ]; then echo yes; else echo no; fi; }

contains() {
  # contains <file> <literal string>
  if grep -qF -- "$2" "$1"; then echo yes; else echo no; fi
}

cleanup() {
  [ -n "${work_root}" ] && [ -d "${work_root}" ] && rm -rf "${work_root}"
}
trap cleanup EXIT INT TERM

resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

python_bin="$(resolve_tool python3 "${PYTHON:-}")" || {
  echo "FAIL: python3 not found on PATH." >&2
  echo "  Install Python 3.11 or newer and try again." >&2
  exit 1
}

work_root="$(mktemp -d "${TMPDIR:-/tmp}/day090-XXXXXX")"
work="${work_root}/lab"
out="${work_root}/out"
mkdir -p "${work}" "${out}"
cp "${lab_dir}/examples/"*.py "${work}/"

echo "Day 090 — A Real Data Layer"
echo

# ===========================================================================
echo "1. The interpreter reports itself, and the module is the standard one"
# ===========================================================================
py_version="$("${python_bin}" -c 'import sys; print(".".join(map(str, sys.version_info[:3])))')"
sqlite_version="$("${python_bin}" -c 'import sqlite3; print(sqlite3.sqlite_version)')"
echo "     python:                 ${py_version}"
echo "     sqlite3.sqlite_version: ${sqlite_version}"

"${python_bin}" -c 'import sqlite3, sys; sys.exit(0 if sqlite3.paramstyle == "qmark" else 1)'
check "the sqlite3 module is present and its paramstyle is qmark" "$(yesno $?)"

"${python_bin}" - <<'PY' >/dev/null 2>&1
import sqlite3, sys
sys.exit(0 if hasattr(sqlite3.connect(":memory:"), "autocommit") else 1)
PY
check "Connection.autocommit exists on this interpreter (Python 3.12 or newer)" "$(yesno $?)"

"${python_bin}" -c 'import sqlite3, sys; sys.exit(0 if sqlite3.sqlite_version_info >= (3, 37, 0) else 1)'
check "the linked SQLite is 3.37.0 or newer, so STRICT tables are available" "$(yesno $?)"

# ===========================================================================
echo
echo "2. The connection factory configures what has to be configured"
# ===========================================================================
cd "${work}" || exit 1

"${python_bin}" - <<'PY' >"${out}/factory.txt" 2>&1
import sqlite3, tempfile, shutil
from pathlib import Path
from db import connect, apply_schema

sandbox = Path(tempfile.mkdtemp(prefix="day090-factory-"))
try:
    path = sandbox / "x.db"
    configured = connect(path)
    apply_schema(configured)
    print("foreign_keys(configured):", configured.execute("PRAGMA foreign_keys").fetchone()[0])
    print("isolation_level:", repr(configured.isolation_level))
    row = configured.execute("SELECT 1 AS one").fetchone()
    print("row type:", type(row).__name__, "by name:", row["one"])
    configured.close()

    raw = sqlite3.connect(path)
    print("foreign_keys(raw):", raw.execute("PRAGMA foreign_keys").fetchone()[0])
    raw.close()
finally:
    shutil.rmtree(sandbox, ignore_errors=True)
PY
check "the factory turns foreign keys ON" \
  "$(contains "${out}/factory.txt" 'foreign_keys(configured): 1')"
check "a plain sqlite3.connect leaves them OFF — the setting is per connection" \
  "$(contains "${out}/factory.txt" 'foreign_keys(raw): 0')"
check "the factory turns off the module's implicit transaction handling" \
  "$(contains "${out}/factory.txt" 'isolation_level: None')"
check "rows arrive as sqlite3.Row, addressable by column name" \
  "$(contains "${out}/factory.txt" 'row type: Row by name: 1')"

# ===========================================================================
echo
echo "3. Injection: the same value as CODE, then as DATA"
# ===========================================================================
"${python_bin}" injection_demo.py >"${out}/injection.txt" 2>&1
injection_status=$?
check "injection_demo.py runs and every one of its own assertions holds" "$(yesno ${injection_status})"
check "the concatenated query leaked all three private rows" \
  "$(contains "${out}/injection.txt" 'rows: 3  -> every member, with address and PIN')"
check "the crafted value changed the statement's meaning" \
  "$(contains "${out}/injection.txt" "WHERE name = 'Ada' OR '1'='1'")"
check "execute() refuses a second statement — a module limit, not a defence" \
  "$(contains "${out}/injection.txt" 'ProgrammingError: You can only execute one statement at a time.')"
check "executescript() accepts it, and the table is destroyed for real" \
  "$(contains "${out}/injection.txt" 'members table exists afterwards: False')"
check "the identical value, bound, returns zero rows" \
  "$(contains "${out}/injection.txt" 'rows returned: 0')"
check "and leaves all three members in place" \
  "$(contains "${out}/injection.txt" 'ok: and still holds all three rows')"
check "the demonstration built its database inside a throwaway directory" \
  "$(contains "${out}/injection.txt" 'sandbox removed. Nothing outside it was ever opened.')"

# The repository must never be fooled by the same input.
"${python_bin}" - <<'PY' >"${out}/repo_injection.txt" 2>&1
import shutil, tempfile
from pathlib import Path
import seed
from db import BookRepository

sandbox = Path(tempfile.mkdtemp(prefix="day090-repoinj-"))
try:
    connection = seed.build(sandbox / "library.db")
    books = BookRepository(connection)
    before = books.count()
    for hostile in ("Fred Brooks' OR '1'='1", "Fred Brooks'; DROP TABLE books; --"):
        found = books.find_by_author(hostile)
        print(f"rows for hostile input: {len(found)}")
    print("books before:", before, "after:", books.count())
    connection.close()
finally:
    shutil.rmtree(sandbox, ignore_errors=True)
PY
check "the repository returns nothing for either hostile author name" \
  "$(contains "${out}/repo_injection.txt" 'rows for hostile input: 0')"
check "and the books table is unchanged afterwards" \
  "$(contains "${out}/repo_injection.txt" 'books before: 7 after: 7')"

# ===========================================================================
echo
echo "4. No SQL anywhere in this lab is built out of pieces"
# ===========================================================================
"${python_bin}" no_sql_strings.py "${work}" >"${out}/guard.txt" 2>&1
check "the guard finds no assembled SQL in the lab's own code" "$(yesno $?)"

guard_dir="${work_root}/guardcheck"
mkdir -p "${guard_dir}"
cat >"${guard_dir}/bad.py" <<'PY'
def find(connection, author):
    return connection.execute(f"SELECT * FROM books WHERE author = '{author}'").fetchall()
PY
"${python_bin}" no_sql_strings.py "${guard_dir}" >"${out}/guard_bad.txt" 2>&1
guard_bad_status=$?
if [ ${guard_bad_status} -ne 0 ]; then bad_ok=0; else bad_ok=1; fi
check "and the guard does catch a deliberately unsafe f-string" "$(yesno ${bad_ok})"
check "naming the file, the line and the reason" \
  "$(contains "${out}/guard_bad.txt" 'bad.py:2: SQL built by an f-string and passed straight to execute()')"

cat >"${guard_dir}/fine.py" <<'PY'
def find(connection, author):
    return connection.execute(
        "SELECT book_id, title FROM books"
        " WHERE author = ? ORDER BY year",
        (author,),
    ).fetchall()
PY
rm -f "${guard_dir}/bad.py"
"${python_bin}" no_sql_strings.py "${guard_dir}" >/dev/null 2>&1
check "and does NOT flag two adjacent string literals, which Python joins at compile time" \
  "$(yesno $?)"

# ===========================================================================
echo
echo "5. Transactions: all of it, or none of it"
# ===========================================================================
"${python_bin}" transactions_demo.py >"${out}/transactions.txt" 2>&1
check "transactions_demo.py runs to the end" "$(yesno $?)"
check "a fresh connection defaults to implicit transaction handling" \
  "$(contains "${out}/transactions.txt" "default isolation_level: ''")"
check "DDL opens no transaction; DML does" \
  "$(contains "${out}/transactions.txt" 'after INSERT (DML):                   True')"
check "with connection: rolls back when the block raises" \
  "$(contains "${out}/transactions.txt" 'rows in t now: [1]')"
check "with connection: does NOT close the connection" \
  "$(contains "${out}/transactions.txt" 'is the connection still usable after the with-block? True')"
check "a foreign-key failure undid the earlier write in the same transaction" \
  "$(contains "${out}/transactions.txt" 'the transaction raised: IntegrityError: FOREIGN KEY constraint failed')"
check "PRAGMA foreign_keys set inside a transaction is silently ignored" \
  "$(contains "${out}/transactions.txt" 'inside a transaction, set ON   -> 0')"
check "and takes effect outside one" \
  "$(contains "${out}/transactions.txt" 'outside again, set ON          -> 1')"
check "autocommit = True makes a write visible to another connection at once" \
  "$(contains "${out}/transactions.txt" 'autocommit = True:  in_transaction False')"

# The rollback property, asserted directly rather than read from a log.
"${python_bin}" - <<'PY' >"${out}/rollback.txt" 2>&1
import shutil, sqlite3, sys, tempfile
from pathlib import Path
import seed
from db import BookRepository, LoanRepository, connect, transaction

sandbox = Path(tempfile.mkdtemp(prefix="day090-rollback-"))
ok = True
try:
    path = sandbox / "library.db"
    connection = seed.build(path)
    books, loans = BookRepository(connection), LoanRepository(connection)
    before = (books.get(1).copies, loans.open_count(), books.count())

    try:
        with transaction(connection):
            loans.borrow(1, 1, "2026-08-01", "2026-08-15")
            loans.borrow(2, 999, "2026-08-01", "2026-08-15")
    except sqlite3.IntegrityError:
        pass
    after_sql = (books.get(1).copies, loans.open_count(), books.count())

    try:
        with transaction(connection):
            loans.borrow(1, 1, "2026-08-01", "2026-08-15")
            raise ZeroDivisionError("a bug in the middle of a transaction")
    except ZeroDivisionError:
        pass
    after_python = (books.get(1).copies, loans.open_count(), books.count())
    connection.close()

    # A second connection sees the same thing: nothing was committed.
    fresh = connect(path)
    after_reopen = (
        BookRepository(fresh).get(1).copies,
        LoanRepository(fresh).open_count(),
        BookRepository(fresh).count(),
    )
    fresh.close()

    print("before:      ", before)
    print("after sql:   ", after_sql)
    print("after python:", after_python)
    print("after reopen:", after_reopen)
    ok = before == after_sql == after_python == after_reopen
    print("UNCHANGED:", ok)
finally:
    shutil.rmtree(sandbox, ignore_errors=True)
sys.exit(0 if ok else 1)
PY
check "a SQL error and a Python error both leave the database byte-identical" "$(yesno $?)"
check "and a newly opened connection agrees that nothing was committed" \
  "$(contains "${out}/rollback.txt" 'UNCHANGED: True')"

# ===========================================================================
echo
echo "6. Cursors, fetch methods and row factories"
# ===========================================================================
"${python_bin}" cursors_demo.py >"${out}/cursors.txt" 2>&1
check "cursors_demo.py runs to the end" "$(yesno $?)"
check "execute() returns a Cursor" \
  "$(contains "${out}/cursors.txt" 'type(connection.execute(...)) -> Cursor')"
check "rowcount is -1 for a SELECT, because the row count is not known in advance" \
  "$(contains "${out}/cursors.txt" 'cursor.rowcount for a SELECT -> -1')"
check "an exhausted cursor returns None from fetchone" \
  "$(contains "${out}/cursors.txt" 'fetchone() now  -> None')"
check "sqlite3.Row is not a dict" \
  "$(contains "${out}/cursors.txt" 'isinstance(row, dict) = False')"
check "and a dict factory produces one when a dict is what you need" \
  "$(contains "${out}/cursors.txt" 'type: dict')"

"${python_bin}" - <<'PY' >"${out}/memory.txt" 2>&1
import shutil, sys, tempfile, tracemalloc
from pathlib import Path
import seed
from db import BookRepository, transaction

sandbox = Path(tempfile.mkdtemp(prefix="day090-memory-"))
try:
    connection = seed.build(sandbox / "library.db")
    connection.execute("CREATE TABLE wide (n INTEGER, payload TEXT) STRICT")
    with transaction(connection):
        connection.executemany(
            "INSERT INTO wide (n, payload) VALUES (?, ?)",
            [(n, "x" * 200) for n in range(40_000)],
        )
    tracemalloc.start()
    rows = connection.execute("SELECT n, payload FROM wide").fetchall()
    peak_fetchall = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    del rows

    tracemalloc.start()
    total = 0
    for row in connection.execute("SELECT n, payload FROM wide"):
        total += row["n"]
    peak_iterate = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    connection.close()
    print(f"fetchall peak: {peak_fetchall:,}  iterate peak: {peak_iterate:,}")
    sys.exit(0 if peak_fetchall > peak_iterate * 100 else 1)
finally:
    shutil.rmtree(sandbox, ignore_errors=True)
PY
check "fetchall holds the whole result in memory; iterating a cursor does not (>100x apart)" \
  "$(yesno $?)"

# ===========================================================================
echo
echo "7. Errors: each mistake raises the class it should"
# ===========================================================================
"${python_bin}" errors_demo.py >"${out}/errors.txt" 2>&1
check "errors_demo.py made thirteen deliberate mistakes and all thirteen raised" "$(yesno $?)"
check "a broken constraint raises IntegrityError" \
  "$(contains "${out}/errors.txt" 'duplicate title (UNIQUE)                               IntegrityError')"
check "a missing table raises OperationalError" \
  "$(contains "${out}/errors.txt" 'table that is not there                                OperationalError')"
check "misusing the module raises ProgrammingError" \
  "$(contains "${out}/errors.txt" 'two statements in one execute()                        ProgrammingError')"
check "a STRICT column refuses the wrong type" \
  "$(contains "${out}/errors.txt" 'cannot store TEXT value in INTEGER column books.year')"

# ===========================================================================
echo
echo "8. executemany, and the transaction that matters more"
# ===========================================================================
"${python_bin}" bulk_insert.py 2000 >"${out}/bulk.txt" 2>&1
check "bulk_insert.py stored every row by all three methods" "$(yesno $?)"
"${python_bin}" - "${out}/bulk.txt" <<'PY' >"${out}/bulk_order.txt" 2>&1
import re, sys
text = open(sys.argv[1], encoding="utf-8").read()
seconds = {}
for line in text.splitlines():
    match = re.match(r"^(.+?)\s{2,}(\d+\.\d+)\s+([\d,]+)\s+([\d.]+)x$", line)
    if match:
        seconds[match.group(1).strip()] = float(match.group(2))
print(seconds)
loop = seconds["a loop, no transaction"]
batched = seconds["a loop inside one transaction"]
many = seconds["executemany inside one transaction"]
print(f"loop/batched = {loop / batched:.1f}x   batched/many = {batched / many:.2f}x")
sys.exit(0 if loop > batched > 0 and many > 0 else 1)
PY
check "one transaction around the loop is dramatically faster than one per row" "$(yesno $?)"
check "all three methods were timed and reported" \
  "$(contains "${out}/bulk.txt" 'executemany inside one transaction')"

# ===========================================================================
echo
echo "9. The data layer's own suite, and the boundary it protects"
# ===========================================================================
"${python_bin}" test_repository.py >"${out}/unittest.txt" 2>&1
check "python3 test_repository.py exits 0" "$(yesno $?)"
check "and reports OK" "$(contains "${out}/unittest.txt" 'OK')"
unit_count="$(grep -oE '^Ran [0-9]+ test' "${out}/unittest.txt" | grep -oE '[0-9]+' | head -1)"
echo "     unit tests run: ${unit_count:-0}"
if [ "${unit_count:-0}" -ge 25 ]; then units_ok=0; else units_ok=1; fi
check "the suite contains at least 25 tests" "$(yesno ${units_ok})"

"${python_bin}" report.py >"${out}/report.txt" 2>&1
check "report.py runs the whole application layer" "$(yesno $?)"
check "the three-table overdue report is right" \
  "$(contains "${out}/report.txt" 'The Mythical Man-Month           due 2026-06-22  (55 days late)')"
check "a duplicate title surfaces as a domain error, never as sqlite3.IntegrityError" \
  "$(contains "${out}/report.txt" "refused: a book titled 'Compilers' is already stored")"
check "a sort key that is not on the allow-list is refused before any SQL exists" \
  "$(contains "${out}/report.txt" 'cannot sort by')"

if grep -qE '^\s*import sqlite3|^\s*from sqlite3' "${work}/report.py" "${work}/domain.py"; then
  boundary=1
else
  boundary=0
fi
check "neither the application layer nor the domain imports sqlite3" "$(yesno ${boundary})"

# ===========================================================================
echo
echo "10. The starter cannot look finished before it is"
# ===========================================================================
starter_work="${work_root}/starter"
mkdir -p "${starter_work}"
cp "${lab_dir}/starter/"*.py "${starter_work}/"
( cd "${starter_work}" && "${python_bin}" smoke.py >"${out}/starter.txt" 2>&1 )
starter_status=$?
if [ ${starter_status} -ne 0 ]; then starter_ok=0; else starter_ok=1; fi
check "the shipped starter exits non-zero and names exercise 1" "$(yesno ${starter_ok})"
check "and says so in words rather than a traceback" \
  "$(contains "${out}/starter.txt" '0 of 9 exercises finished.')"

cp "${work}/db.py" "${starter_work}/db.py"
( cd "${starter_work}" && "${python_bin}" smoke.py >"${out}/starter_done.txt" 2>&1 )
check "a completed db.py takes the same starter to exit 0" "$(yesno $?)"
check "reporting all nine" \
  "$(contains "${out}/starter_done.txt" '9 of 9 exercises finished.')"

# ===========================================================================
echo
echo "11. Offline, self-contained, and leaves nothing behind"
# ===========================================================================
if grep -rlE 'https?://|[0-9]{1,3}(\.[0-9]{1,3}){3}' "${lab_dir}/examples" "${lab_dir}/starter" >/dev/null 2>&1; then
  net=1
else
  net=0
fi
check "no executable lab file contains a network address of any kind" "$(yesno ${net})"

if grep -rhE '^\s*(import|from)\s+' "${lab_dir}/examples" "${lab_dir}/starter" \
  | grep -qE '\b(requests|httpx|urllib3|pandas|sqlalchemy|aiosqlite|numpy|pytest)\b'; then
  thirdparty=1
else
  thirdparty=0
fi
check "no lab file imports a third-party package — standard library only" "$(yesno ${thirdparty})"

if grep -rq 'sudo' "${lab_dir}/examples" "${lab_dir}/starter"; then sudo_used=1; else sudo_used=0; fi
check "nothing in the lab's code asks for sudo" "$(yesno ${sudo_used})"

leftover="$(find "${lab_dir}" -name '*.db' -o -name '*.db-journal' -o -name '*.db-wal' | wc -l | tr -d ' ')"
if [ "${leftover}" = "0" ]; then left_ok=0; else left_ok=1; fi
check "this run left no database file anywhere inside the lab directory" "$(yesno ${left_ok})"

# Every script makes its own sandbox with a day090- prefix and removes it in
# a finally block. The only one that may still exist is this harness's own,
# which its trap removes when the script exits.
stray="$(find "${TMPDIR:-/tmp}" -maxdepth 1 -name 'day090-*' \
  ! -name "$(basename "${work_root}")" 2>/dev/null | wc -l | tr -d ' ')"
if [ "${stray}" = "0" ]; then stray_ok=0; else stray_ok=1; fi
check "no sandbox from any lab script was left in the temporary directory" "$(yesno ${stray_ok})"

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