Day 6 Checkpoint: Batch and Bound

Two measurements, two changes. The N+1 benchmark says the library needs a batched read; tracemalloc says the identity map needs a limit. Both are yours to write.

The checkpoint

Checkpoint

Day 6 checkpoint: get_many and a bounded cache

models.py has two NotImplementedErrors: _remember, which must evict, and get_many, which must fetch what it doesn’t already have in one query. The tests count queries — there is not a single timing assertion in the file.

models.py Python

"""The base model, typed. `get` returns `Self`, so `Post.get(...)` is a `Post` to a checker, not a `Model`. The connection is a `Protocol`, so the library never imports sqlite3. """ from collections import OrderedDict from collections.abc import Iterable, Sequence from typing import Any, ClassVar, Self, cast from warnings import deprecated from errors import NotFound from fields import Field, IntegerField, TextField from protocols import Connection from transactions import transaction class Model: #: Every model, by table name. registry: ClassVar[dict[str, type["Model"]]] = {} table: ClassVar[str] fields: ClassVar[dict[str, Field[Any]]] #: The identity map, oldest first, capped at cache_size entries. _instances: ClassVar[OrderedDict[int, Any]] #: How many loaded objects a model keeps. Day 6 measured what happens #: without a cap: the cache grows for the life of the process. cache_size: ClassVar[int] = 128 id: int | None def __init_subclass__(cls, table: str | None = None, **kwargs: Any) -> None: super().__init_subclass__(**kwargs) cls.table = table or cls.__name__.lower() cls.fields = { name: value for klass in reversed(cls.__mro__) for name, value in vars(klass).items() if isinstance(value, Field) } cls._instances = OrderedDict() Model.registry[cls.table] = cls def __init__(self, id: int | None = None, **values: Any) -> None: self.id = id for name in self.fields: if name in values: setattr(self, name, values.pop(name)) if values: raise TypeError(f"{type(self).__name__} has no column {sorted(values)[0]!r}") @property def columns(self) -> tuple[str, ...]: return tuple(self.fields) def save(self, connection: Connection) -> Self: """Write this object. All of it, or none of it.""" with transaction(connection): self._write(connection) self._remember() # only once the write is committed return self @classmethod def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None: """Write several objects in one transaction: all of them, or none.""" written = list(objects) with transaction(connection): for obj in written: obj._write(connection) for obj in written: obj._remember() def _write(self, connection: Connection) -> None: columns = self.columns values = [getattr(self, name) for name in columns] names = ", ".join(columns) if self.id is None: marks = ", ".join("?" for _ in columns) cursor = connection.execute( f"insert into {self.table} ({names}) values ({marks})", values ) self.id = cursor.lastrowid else: assignments = ", ".join(f"{name} = ?" for name in columns) connection.execute( f"update {self.table} set {assignments} where id = ?", [*values, self.id] ) def _remember(self) -> None: """Put this object in its model's identity map, evicting the oldest.""" assert self.id is not None model = type(self) cache = model._instances # TODO: put self in the cache, mark it most recently used, and drop # the least recently used entries while the cache is over # model.cache_size. OrderedDict has move_to_end and popitem(last=False). raise NotImplementedError("_remember") @classmethod def get(cls, connection: Connection, id: int) -> Self: cached = cls._instances.get(id) if cached is not None: cls._instances.move_to_end(id) return cast(Self, cached) row = connection.execute(f"select * from {cls.table} where id = ?", (id,)).fetchone() if row is None: raise NotFound(f"no {cls.__name__} with id {id}") # Every value goes through the fields, so what the database returns is # validated exactly like what a caller assigns. found = cls(**{key: row[key] for key in row.keys()}) found._remember() return found @classmethod def get_many(cls, connection: Connection, ids: Sequence[int]) -> dict[int, Self]: """Every row in one query, instead of one query per row. Anything already in the identity map is served from there, so a second call for the same ids costs nothing. """ # TODO: one query for everything that isn't already cached. # - de-duplicate the ids, keeping the order asked for # - serve what the identity map already holds # - fetch the rest with a single "... where id in (?, ?, ...)" # - build each row through cls(**...), so the fields validate it # - return {id: object} in the order the caller asked for, leaving # out ids that do not exist raise NotImplementedError("get_many") @classmethod @deprecated("Model.fetch is deprecated; use Model.get instead") def fetch(cls, connection: Connection, id: int) -> Self: return cls.get(connection, id) def __repr__(self) -> str: shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields) return f"{type(self).__name__}(id={self.id!r}, {shown})" class Post(Model, table="post"): title = TextField(max_length=100) views = IntegerField(minimum=0) class User(Model, table="user"): email = TextField(max_length=200)

"""A connection wrapper that counts what the library asks the database to do. Timings vary with the machine; counts don't. Most performance bugs in a data library are "how many queries", and that is worth asserting in a test. """ from collections.abc import Sequence from typing import Any from protocols import Connection, Cursor class CountingConnection: """Passes everything through to a real connection, and counts it.""" def __init__(self, connection: Connection) -> None: self.real = connection self.queries: list[str] = [] def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: self.queries.append(sql) return self.real.execute(sql, parameters) def commit(self) -> None: self.real.commit() @property def selects(self) -> int: return sum(1 for sql in self.queries if sql.lstrip().startswith("select")) def reset(self) -> None: self.queries.clear()

"""A workload big enough to measure, and small enough to run in a browser. python benchmark.py the numbers below python -m cProfile -s tottime benchmark.py where the whole run's time goes Every result is the best of several runs, with the cache reset before each, as the measuring lesson recommends. The numbers are illustrative — the 0.2 ms is a simulated stand-in for a round trip, not what a real query costs — and both the timings and the ratio change on different hardware. What is stable is the shape: without latency, 2000 queries and 1 query cost about the same; with latency, they do not. To profile the isolated one-by-one path (not setup or both latency modes), import from a REPL and profile just that call: python -c "import benchmark, cProfile; \\ c = benchmark.connect(); ids = benchmark.fill(c); \\ benchmark.Post._instances.clear(); \\ cProfile.run('benchmark.one_by_one(c, ids)', sort='tottime')" """ import time from collections.abc import Sequence from typing import Any from counting import CountingConnection from db import connect from models import Post ROWS = 2_000 LATENCY = 0.0002 # a SIMULATED 0.2 ms per query, not a real round trip RUNS = 5 class LatentConnection(CountingConnection): """A connection with a simulated per-query delay.""" def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Any: time.sleep(LATENCY) return super().execute(sql, parameters) def fill(connection: Any) -> list[int]: Post.save_all(connection, [Post(title=f"post {index}", views=index) for index in range(ROWS)]) return [index + 1 for index in range(ROWS)] def one_by_one(connection: Any, ids: Sequence[int]) -> list[Post]: return [Post.get(connection, id) for id in ids] def in_one_query(connection: Any, ids: Sequence[int]) -> list[Post]: return list(Post.get_many(connection, ids).values()) def best_of(label: str, function: Any, connection: Any, ids: Sequence[int], wrapper: type[CountingConnection]) -> float: """Best of RUNS, cache reset before each so every run starts cold.""" best = float("inf") selects = 0 for _ in range(RUNS): Post._instances.clear() counter = wrapper(connection) start = time.perf_counter() function(counter, ids) best = min(best, time.perf_counter() - start) selects = counter.selects print(f" {label:12} {selects:6} selects {best * 1000:9.1f} ms (best of {RUNS})") return best def main() -> None: connection = connect() ids = fill(connection) print(f"{ROWS} rows, cache_size={Post.cache_size}\n") print("in-memory SQLite (no latency):") slow = best_of("one by one", one_by_one, connection, ids, CountingConnection) fast = best_of("get_many", in_one_query, connection, ids, CountingConnection) print(f" ratio: {slow / fast:.1f}x\n") print(f"with {LATENCY * 1000:.1f} ms of SIMULATED latency per query:") slow = best_of("one by one", one_by_one, connection, ids, LatentConnection) fast = best_of("get_many", in_one_query, connection, ids, LatentConnection) print(f" ratio: {slow / fast:.0f}x") if __name__ == "__main__": main()

"""Measure the identity map with tracemalloc: bounded versus effectively not. python leak.py Loads the same 5000 rows one at a time under two cache sizes and reports how much memory survived the load, and where it was allocated. This is the script that produced the numbers in Day 6's memory lesson. """ import tracemalloc from db import connect from models import Post ROWS = 5000 def load(connection, ids): for id in ids: Post.get(connection, id) def measure(connection, ids, cache_size): Post.cache_size = cache_size Post._instances.clear() tracemalloc.start() before = tracemalloc.take_snapshot() load(connection, ids) after = tracemalloc.take_snapshot() tracemalloc.stop() grew = sum(stat.size_diff for stat in after.compare_to(before, "lineno")) print(f"cache_size={cache_size:6}: cache holds {len(Post._instances):5} objects, " f"memory grew {grew / 1024:8.0f} KiB") for stat in after.compare_to(before, "lineno")[:3]: where = str(stat.traceback[0]) print(f" {where:40} {stat.size_diff / 1024:8.0f} KiB") def main(): connection = connect() Post.save_all(connection, [Post(title=f"post {index}", views=index) for index in range(ROWS)]) ids = [index + 1 for index in range(ROWS)] measure(connection, ids, cache_size=10_000) # effectively unbounded here measure(connection, ids, cache_size=128) # the shipped default if __name__ == "__main__": main()

"""Fields, typed so a checker knows what each column holds. `Field` is generic in the value it stores, so `IntegerField` is a `Field[int]` and a checker can tell that `post.views` is an `int` — without any annotation in the model itself. """ from typing import Any, Self, cast, overload from errors import ValidationError class Field[T]: """A column of type T, validated on assignment.""" name: str slot: str def __init__(self, *, null: bool = False) -> None: self.null = null self.name = "<unnamed>" self.slot = "<unnamed>" def __set_name__(self, owner: type, name: str) -> None: self.name = name self.slot = f"_{name}" # Read through the class (Post.views) you get the field; through an # instance (post.views) you get the value. Two overloads say exactly that. @overload def __get__(self, instance: None, owner: type | None = None) -> Self: ... @overload def __get__(self, instance: object, owner: type | None = None) -> T: ... def __get__(self, instance: object | None, owner: type | None = None) -> Self | T: if instance is None: return self try: return cast(T, instance.__dict__[self.slot]) except KeyError: raise AttributeError(f"{type(instance).__name__}.{self.name} is not set") from None def __set__(self, instance: object, value: T) -> None: instance.__dict__[self.slot] = self.check(value) def check(self, value: Any) -> T: """Return the value to store, or raise ValidationError. `cast` is a promise to the checker, not a conversion: subclasses decide what T really is, and only they can prove it. """ if value is None and not self.null: raise ValidationError(f"{self.name} may not be None") return cast(T, value) def __repr__(self) -> str: return f"{type(self).__name__}({self.name!r})" class TextField(Field[str]): """A text column. Note what the annotation does *not* say: with `null=True` the value may be None at run time, and `Field[str]` still reports `str`. See "What the checker can't see" in the lesson. """ def __init__(self, *, max_length: int | None = None, **kwargs: Any) -> None: super().__init__(**kwargs) self.max_length = max_length def check(self, value: Any) -> str: value = super().check(value) if value is None: return cast(str, None) if not isinstance(value, str): raise ValidationError(f"{self.name} must be a str, not {type(value).__name__}") if self.max_length is not None and len(value) > self.max_length: raise ValidationError(f"{self.name} is longer than {self.max_length} characters") return value class IntegerField(Field[int]): """An integer column. As above, `null=True` is invisible to the checker.""" def __init__(self, *, minimum: int | None = None, **kwargs: Any) -> None: super().__init__(**kwargs) self.minimum = minimum def check(self, value: Any) -> int: value = super().check(value) if value is None: return cast(int, None) if isinstance(value, bool) or not isinstance(value, int): raise ValidationError(f"{self.name} must be an int, not {type(value).__name__}") if self.minimum is not None and value < self.minimum: raise ValidationError(f"{self.name} must be at least {self.minimum}") return value

"""Every exception this library raises, and how driver errors are translated. One base class, `ORMError`, so a caller can catch everything from this library with one except clause. Each specific error also inherits from the built-in it replaces, so code written against the old API keeps working. """ class ORMError(Exception): """Base class for every error raised by this library.""" class ValidationError(ORMError, ValueError): """A value a field refused. Also a ValueError, as it was before.""" class NotFound(ORMError, LookupError): """No row with that id. Also a LookupError, as it was before.""" class IntegrityError(ORMError): """The database refused the write: a constraint, a duplicate key.""" #: Driver exceptions to translate into IntegrityError. Empty until a driver #: registers itself, because the library never imports one. DRIVER_INTEGRITY_ERRORS: tuple[type[BaseException], ...] = () def use_driver(*, integrity: type[BaseException]) -> None: """Tell the library which driver exception means "the database said no". Called by the code that owns the connection (here, db.connect), because only it knows which driver is in use. """ global DRIVER_INTEGRITY_ERRORS DRIVER_INTEGRITY_ERRORS = (integrity,)

"""All-or-nothing writes, including nested ones. The commit is inside the exception handler, not after it: a database can fail *during* commit (a deferred constraint is the classic case), and that failure has to roll back and translate exactly like a failure in the body. """ from collections.abc import Iterator from contextlib import contextmanager import errors from protocols import Connection #: Open transactions, by connection. An entry exists only while a transaction #: is running: sqlite3 connections can be neither weak-referenced nor given #: attributes, and a dict that kept every connection forever would be a leak. _depth: dict[Connection, int] = {} def _rollback(connection: Connection, depth: int, savepoint: str) -> None: if depth == 0: connection.execute("rollback") else: connection.execute(f"rollback to {savepoint}") connection.execute(f"release {savepoint}") def _finish(connection: Connection, depth: int, savepoint: str) -> None: """Commit the outermost transaction, or release an inner savepoint. This can raise — a commit is when deferred constraints are checked — so it lives inside the try, and its failure is handled like any other. """ if depth == 0: connection.execute("commit") else: connection.execute(f"release {savepoint}") def _translate(error: BaseException) -> None: """Turn a driver integrity error into ours; leave everything else alone. Notes ride along. Anything the failed rollback wrote onto the original is copied to the replacement: a translation that silently dropped notes would throw away the diagnostic it was worth attaching them for. """ if isinstance(error, errors.DRIVER_INTEGRITY_ERRORS): ours = errors.IntegrityError(str(error)) for note in getattr(error, "__notes__", ()): ours.add_note(note) raise ours from error @contextmanager def transaction(connection: Connection) -> Iterator[Connection]: """Commit on success, roll back on any exception. Nesting is allowed: an inner `with` becomes a savepoint, so it can fail and be undone without throwing away the work the outer one has done. A failure while committing is rolled back too. Policy on a *failed* rollback: the original error is what the caller needs, so a second error from the rollback itself never replaces it — it is attached to the first as a note. This is the common case for a driver conflict clause (SQLite's ``INSERT OR ROLLBACK``) that has already ended the transaction. It is not a promise to repair a genuinely broken connection: a connection left unusable by such a failure stays that way, and the note is how you find that out from the traceback rather than by guessing. """ depth = _depth.get(connection, 0) savepoint = f"orm_{depth}" connection.execute("begin" if depth == 0 else f"savepoint {savepoint}") _depth[connection] = depth + 1 try: try: yield connection _finish(connection, depth, savepoint) # commit/release can fail except BaseException as error: try: _rollback(connection, depth, savepoint) except Exception as cleanup_error: # A conflict clause may have rolled back already, so this is # usually noise: keep the primary error. But write down what # the cleanup hit, so a genuinely broken connection leaves # evidence instead of vanishing quietly. error.add_note( f"rollback after this error also failed: {cleanup_error!r}") _translate(error) # raises ours, or returns raise finally: if depth: _depth[connection] = depth else: _depth.pop(connection, None) # nothing is kept between transactions

"""What the library needs from a database connection — and nothing more. Typing these as `sqlite3.Connection` would tie the library to one driver. A Protocol says what it *uses*, so any object with those methods fits: a real connection, a wrapper that logs, a fake in a test. """ from collections.abc import Sequence from typing import Any, Protocol, runtime_checkable class Row(Protocol): """A result row that can be turned into a dict of column values.""" def keys(self) -> Sequence[str]: ... def __getitem__(self, key: str, /) -> Any: ... @runtime_checkable class Cursor(Protocol): # A property, not a plain attribute: an attribute in a Protocol means # read *and* write, and sqlite3's lastrowid is read-only. @property def lastrowid(self) -> int | None: ... def fetchone(self) -> Row | None: ... def fetchall(self) -> Sequence[Row]: ... @runtime_checkable class Connection(Protocol): def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: ... def commit(self) -> None: ...

"""SQLite scaffolding, supplied by the course. `isolation_level=None` turns off the driver's own implicit transactions, so the library's `transaction()` is the only thing issuing begin, commit and rollback. """ import sqlite3 import errors SCHEMA = """ create table if not exists post ( id integer primary key autoincrement, title text not null unique, views integer not null default 0 ); create table if not exists user ( id integer primary key autoincrement, email text not null ); """ def connect(path: str = ":memory:") -> sqlite3.Connection: connection = sqlite3.connect(path, isolation_level=None) connection.row_factory = sqlite3.Row connection.executescript(SCHEMA) errors.use_driver(integrity=sqlite3.IntegrityError) return connection

import unittest import transactions from counting import CountingConnection from db import connect from models import Model, Post class Queries(unittest.TestCase): def setUp(self) -> None: self.real = connect() for model in Model.registry.values(): model._instances.clear() Post.save_all(self.real, [Post(title=f"post {index}", views=index) for index in range(20)]) for model in Model.registry.values(): model._instances.clear() self.connection = CountingConnection(self.real) def test_one_by_one_costs_one_query_each(self) -> None: for id in range(1, 11): Post.get(self.connection, id) self.assertEqual(self.connection.selects, 10) def test_get_many_costs_one_query(self) -> None: Post.get_many(self.connection, list(range(1, 11))) self.assertEqual(self.connection.selects, 1) def test_get_many_returns_what_was_asked_for_in_order(self) -> None: found = Post.get_many(self.connection, [3, 1, 2]) self.assertEqual(list(found), [3, 1, 2]) self.assertEqual([post.title for post in found.values()], ["post 2", "post 0", "post 1"]) def test_repeated_ids_are_asked_for_once(self) -> None: Post.get_many(self.connection, [1, 1, 2, 2, 1]) self.assertEqual(self.connection.selects, 1) self.assertEqual(self.connection.queries[-1].count("?"), 2) def test_cached_objects_cost_nothing(self) -> None: Post.get_many(self.connection, [1, 2, 3]) self.connection.reset() again = Post.get_many(self.connection, [1, 2, 3]) self.assertEqual(self.connection.selects, 0) self.assertEqual(len(again), 3) def test_a_partially_cached_request_only_fetches_the_rest(self) -> None: Post.get_many(self.connection, [1, 2]) self.connection.reset() Post.get_many(self.connection, [1, 2, 3, 4]) self.assertEqual(self.connection.selects, 1) self.assertEqual(self.connection.queries[-1].count("?"), 2, "only the missing ids") def test_missing_ids_are_simply_absent(self) -> None: found = Post.get_many(self.connection, [1, 999]) self.assertEqual(list(found), [1]) def test_the_same_object_comes_back_each_time(self) -> None: first = Post.get(self.connection, 1) self.assertIs(Post.get_many(self.connection, [1])[1], first) class CacheSize(unittest.TestCase): def setUp(self) -> None: self.connection = connect() for model in Model.registry.values(): model._instances.clear() self.addCleanup(setattr, Post, "cache_size", Post.cache_size) def test_the_cache_does_not_grow_without_limit(self) -> None: Post.cache_size = 10 Post.save_all(self.connection, [Post(title=f"post {index}", views=index) for index in range(200)]) self.assertLessEqual(len(Post._instances), 10) def test_loading_many_rows_does_not_grow_it_either(self) -> None: Post.cache_size = 10 Post.save_all(self.connection, [Post(title=f"post {index}", views=index) for index in range(200)]) Post._instances.clear() for id in range(1, 201): Post.get(self.connection, id) self.assertLessEqual(len(Post._instances), 10) def test_the_oldest_entry_is_the_one_dropped(self) -> None: Post.cache_size = 3 Post.save_all(self.connection, [Post(title=f"post {index}", views=index) for index in range(5)]) Post._instances.clear() for id in (1, 2, 3): Post.get(self.connection, id) Post.get(self.connection, 1) # 1 is now the most recently used Post.get(self.connection, 4) # evicts 2, the least recent self.assertEqual(sorted(Post._instances), [1, 3, 4]) def test_a_cached_object_is_still_the_same_object(self) -> None: Post.cache_size = 100 post = Post(title="kept", views=1).save(self.connection) self.assertIs(Post.get(self.connection, post.id), post) class StillWorks(unittest.TestCase): def setUp(self) -> None: self.connection = connect() for model in Model.registry.values(): model._instances.clear() def test_transactions_leave_nothing_behind(self) -> None: Post(title="one", views=1).save(self.connection) self.assertEqual(transactions._depth, {}) def test_save_all_is_still_atomic(self) -> None: Post(title="one", views=1).save(self.connection) with self.assertRaises(Exception): Post.save_all(self.connection, [Post(title="two", views=1), Post(title="one", views=2)]) titles = [row["title"] for row in self.connection.execute("select title from post")] self.assertEqual(titles, ["one"])

import sys import unittest import test_models suite = unittest.defaultTestLoader.loadTestsFromModule(test_models) unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

Hint 1: the idea

Two halves of one promise: bounded memory and counted queries. _remember is this lesson’s LRU in a new setting — store, mark as most recent, evict the oldest while over cache_size. get_many serves what the identity map already holds and fetches everything else in one where id in (...) query, then answers in the order the caller asked.

Hint 2: a first step

Write _remember first, following its TODO — it’s the LRU put — and run: the cache tests should pass. For get_many, de-duplicate the ids, split them into cached and missing, and build one ? per missing id with ", ".join("?" for _ in missing).

Show one solution

    def _remember(self) -> None:
        """Put this object in its model's identity map, evicting the oldest."""
        assert self.id is not None
        model = type(self)
        cache = model._instances
        cache[self.id] = self
        cache.move_to_end(self.id)
        while len(cache) > model.cache_size:
            cache.popitem(last=False)          # drop the least recently used

    @classmethod
    def get_many(cls, connection: Connection, ids: Sequence[int]) -> dict[int, Self]:
        wanted = list(dict.fromkeys(ids))                    # de-duplicate, keep order
        found: dict[int, Self] = {}
        missing = []
        for id in wanted:
            cached = cls._instances.get(id)
            if cached is None:
                missing.append(id)
            else:
                cls._instances.move_to_end(id)
                found[id] = cast(Self, cached)
        if missing:
            marks = ", ".join("?" for _ in missing)
            rows = connection.execute(
                f"select * from {cls.table} where id in ({marks})", missing
            ).fetchall()
            for row in rows:
                obj = cls(**{key: row[key] for key in row.keys()})
                obj._remember()
                found[cast(int, obj.id)] = obj
        return {id: found[id] for id in wanted if id in found}

Two details the tests force and the code doesn’t advertise.

get_many routes every row through cls(**...) and _remember() — the same path get() uses. It would be a little faster to skip both, and that is exactly the fast path from Day 3’s bug hunt: skip the constructor and you skip the validation, so a text value in an integer column reaches the caller. Batching must not become a second, less careful way in.

while len(cache) > model.cache_size reads model.cache_size, not Model.cache_size, so a model can set its own:

class Post(Model, table="post"):
    cache_size = 1_000

Class attributes resolved through type(self) give you per-model configuration for free — the mechanism from Day 1, used deliberately this time.

Bug hunt: the leak that has been there all week

This version is the library as it stood before today: _remember without the eviction.

Reproduce Python

"""The base model, typed. `get` returns `Self`, so `Post.get(...)` is a `Post` to a checker, not a `Model`. The connection is a `Protocol`, so the library never imports sqlite3. """ from collections import OrderedDict from collections.abc import Iterable, Sequence from typing import Any, ClassVar, Self, cast from warnings import deprecated from errors import NotFound from fields import Field, IntegerField, TextField from protocols import Connection from transactions import transaction class Model: #: Every model, by table name. registry: ClassVar[dict[str, type["Model"]]] = {} table: ClassVar[str] fields: ClassVar[dict[str, Field[Any]]] #: The identity map, oldest first, capped at cache_size entries. _instances: ClassVar[OrderedDict[int, Any]] #: How many loaded objects a model keeps. Day 6 measured what happens #: without a cap: the cache grows for the life of the process. cache_size: ClassVar[int] = 128 id: int | None def __init_subclass__(cls, table: str | None = None, **kwargs: Any) -> None: super().__init_subclass__(**kwargs) cls.table = table or cls.__name__.lower() cls.fields = { name: value for klass in reversed(cls.__mro__) for name, value in vars(klass).items() if isinstance(value, Field) } cls._instances = OrderedDict() Model.registry[cls.table] = cls def __init__(self, id: int | None = None, **values: Any) -> None: self.id = id for name in self.fields: if name in values: setattr(self, name, values.pop(name)) if values: raise TypeError(f"{type(self).__name__} has no column {sorted(values)[0]!r}") @property def columns(self) -> tuple[str, ...]: return tuple(self.fields) def save(self, connection: Connection) -> Self: """Write this object. All of it, or none of it.""" with transaction(connection): self._write(connection) self._remember() # only once the write is committed return self @classmethod def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None: """Write several objects in one transaction: all of them, or none.""" written = list(objects) with transaction(connection): for obj in written: obj._write(connection) for obj in written: obj._remember() def _write(self, connection: Connection) -> None: columns = self.columns values = [getattr(self, name) for name in columns] names = ", ".join(columns) if self.id is None: marks = ", ".join("?" for _ in columns) cursor = connection.execute( f"insert into {self.table} ({names}) values ({marks})", values ) self.id = cursor.lastrowid else: assignments = ", ".join(f"{name} = ?" for name in columns) connection.execute( f"update {self.table} set {assignments} where id = ?", [*values, self.id] ) def _remember(self) -> None: """Put this object in its model's identity map, evicting the oldest.""" assert self.id is not None model = type(self) cache = model._instances cache[self.id] = self cache.move_to_end(self.id) @classmethod def get(cls, connection: Connection, id: int) -> Self: cached = cls._instances.get(id) if cached is not None: cls._instances.move_to_end(id) return cast(Self, cached) row = connection.execute(f"select * from {cls.table} where id = ?", (id,)).fetchone() if row is None: raise NotFound(f"no {cls.__name__} with id {id}") # Every value goes through the fields, so what the database returns is # validated exactly like what a caller assigns. found = cls(**{key: row[key] for key in row.keys()}) found._remember() return found @classmethod def get_many(cls, connection: Connection, ids: Sequence[int]) -> dict[int, Self]: """Every row in one query, instead of one query per row. Anything already in the identity map is served from there, so a second call for the same ids costs nothing. """ wanted = list(dict.fromkeys(ids)) # de-duplicate, keep order found: dict[int, Self] = {} missing = [] for id in wanted: cached = cls._instances.get(id) if cached is None: missing.append(id) else: cls._instances.move_to_end(id) found[id] = cast(Self, cached) if missing: marks = ", ".join("?" for _ in missing) rows = connection.execute( f"select * from {cls.table} where id in ({marks})", missing ).fetchall() for row in rows: obj = cls(**{key: row[key] for key in row.keys()}) obj._remember() found[cast(int, obj.id)] = obj return {id: found[id] for id in wanted if id in found} @classmethod @deprecated("Model.fetch is deprecated; use Model.get instead") def fetch(cls, connection: Connection, id: int) -> Self: return cls.get(connection, id) def __repr__(self) -> str: shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields) return f"{type(self).__name__}(id={self.id!r}, {shown})" class Post(Model, table="post"): title = TextField(max_length=100) views = IntegerField(minimum=0) class User(Model, table="user"): email = TextField(max_length=200)

import sys import unittest import test_models suite = unittest.defaultTestLoader.loadTestsFromModule(test_models) unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

Reproduce. Three failures, all in CacheSize:

AssertionError: 200 not less than or equal to 10

The test set cache_size to 10 and saved 200 posts. The cache holds all 200.

Inspect. There is nothing to step through: _remember inserts and never removes. The interesting part is why nobody noticed for five days — every other test in the suite passes, the library is correct, and the benchmark is faster with a bigger cache. A leak makes the happy path look good.

Narrow. The reference count tells the story. Model._instances is a class attribute, alive as long as the class, and it holds a strong reference to every object ever loaded. Those objects can never be collected, no matter what the caller does with them.

Pin. test_the_cache_does_not_grow_without_limit and its two neighbours. Note what they assert — len(Post._instances), not megabytes. Memory assertions are noisy; entry counts are exact, and the count is what causes the megabytes.

Fix. Evict the least recently used entry once the cache is over its size, and mark an entry as used when it’s read.

Say what the fix cost

With a cap, two get() calls for the same row no longer always return the same object. That is a weakened guarantee, and code that relies on is will fail rarely and mysteriously. This is the sort of change that belongs in a changelog, in the class docstring, and in tomorrow’s design write-up.

Where Day 6 leaves the library

The finished Day 6 Python

"""The base model, typed. `get` returns `Self`, so `Post.get(...)` is a `Post` to a checker, not a `Model`. The connection is a `Protocol`, so the library never imports sqlite3. """ from collections import OrderedDict from collections.abc import Iterable, Sequence from typing import Any, ClassVar, Self, cast from warnings import deprecated from errors import NotFound from fields import Field, IntegerField, TextField from protocols import Connection from transactions import transaction class Model: #: Every model, by table name. registry: ClassVar[dict[str, type["Model"]]] = {} table: ClassVar[str] fields: ClassVar[dict[str, Field[Any]]] #: The identity map, oldest first, capped at cache_size entries. _instances: ClassVar[OrderedDict[int, Any]] #: How many loaded objects a model keeps. Day 6 measured what happens #: without a cap: the cache grows for the life of the process. cache_size: ClassVar[int] = 128 id: int | None def __init_subclass__(cls, table: str | None = None, **kwargs: Any) -> None: super().__init_subclass__(**kwargs) cls.table = table or cls.__name__.lower() cls.fields = { name: value for klass in reversed(cls.__mro__) for name, value in vars(klass).items() if isinstance(value, Field) } cls._instances = OrderedDict() Model.registry[cls.table] = cls def __init__(self, id: int | None = None, **values: Any) -> None: self.id = id for name in self.fields: if name in values: setattr(self, name, values.pop(name)) if values: raise TypeError(f"{type(self).__name__} has no column {sorted(values)[0]!r}") @property def columns(self) -> tuple[str, ...]: return tuple(self.fields) def save(self, connection: Connection) -> Self: """Write this object. All of it, or none of it.""" with transaction(connection): self._write(connection) self._remember() # only once the write is committed return self @classmethod def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None: """Write several objects in one transaction: all of them, or none.""" written = list(objects) with transaction(connection): for obj in written: obj._write(connection) for obj in written: obj._remember() def _write(self, connection: Connection) -> None: columns = self.columns values = [getattr(self, name) for name in columns] names = ", ".join(columns) if self.id is None: marks = ", ".join("?" for _ in columns) cursor = connection.execute( f"insert into {self.table} ({names}) values ({marks})", values ) self.id = cursor.lastrowid else: assignments = ", ".join(f"{name} = ?" for name in columns) connection.execute( f"update {self.table} set {assignments} where id = ?", [*values, self.id] ) def _remember(self) -> None: """Put this object in its model's identity map, evicting the oldest.""" assert self.id is not None model = type(self) cache = model._instances cache[self.id] = self cache.move_to_end(self.id) while len(cache) > model.cache_size: cache.popitem(last=False) # drop the least recently used @classmethod def get(cls, connection: Connection, id: int) -> Self: cached = cls._instances.get(id) if cached is not None: cls._instances.move_to_end(id) return cast(Self, cached) row = connection.execute(f"select * from {cls.table} where id = ?", (id,)).fetchone() if row is None: raise NotFound(f"no {cls.__name__} with id {id}") # Every value goes through the fields, so what the database returns is # validated exactly like what a caller assigns. found = cls(**{key: row[key] for key in row.keys()}) found._remember() return found @classmethod def get_many(cls, connection: Connection, ids: Sequence[int]) -> dict[int, Self]: """Every row in one query, instead of one query per row. Anything already in the identity map is served from there, so a second call for the same ids costs nothing. """ wanted = list(dict.fromkeys(ids)) # de-duplicate, keep order found: dict[int, Self] = {} missing = [] for id in wanted: cached = cls._instances.get(id) if cached is None: missing.append(id) else: cls._instances.move_to_end(id) found[id] = cast(Self, cached) if missing: marks = ", ".join("?" for _ in missing) rows = connection.execute( f"select * from {cls.table} where id in ({marks})", missing ).fetchall() for row in rows: obj = cls(**{key: row[key] for key in row.keys()}) obj._remember() found[cast(int, obj.id)] = obj return {id: found[id] for id in wanted if id in found} @classmethod @deprecated("Model.fetch is deprecated; use Model.get instead") def fetch(cls, connection: Connection, id: int) -> Self: return cls.get(connection, id) def __repr__(self) -> str: shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields) return f"{type(self).__name__}(id={self.id!r}, {shown})" class Post(Model, table="post"): title = TextField(max_length=100) views = IntegerField(minimum=0) class User(Model, table="user"): email = TextField(max_length=200)

import sys import unittest import test_models suite = unittest.defaultTestLoader.loadTestsFromModule(test_models) unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

Working on your own machine? Download the Day 6 files, then:

python benchmark.py
python -m cProfile -s tottime benchmark.py

The first prints the 1.0x and 77x numbers from this morning’s lesson on your hardware. The second shows where the time goes when there’s no latency to hide behind.

Optional: what the interpreter actually does

dis.dis(Model.get) prints the bytecode. It’s worth looking at once — attribute lookups become LOAD_ATTR, the f-string becomes FORMAT_VALUE and BUILD_STRING — because it makes concrete why an attribute access is not free. It is almost never worth optimising at that level: the 77x win today came from removing round trips, not instructions.

What you learned

Batch reads through the same construction path as single reads, or batching becomes a way to skip validation.
type(self).cache_size gives per-model configuration through ordinary attribute lookup.
An unbounded cache passes every functional test; only a test that asserts on its size catches it.
Every bound you add weakens a guarantee somewhere — write that down.

Tomorrow: finish the library, test the week’s edge cases, benchmark it, and write the design document that says what all of these trade-offs were.

The checkpoint

Bug hunt: the leak that has been there all week

Say what the fix cost

Where Day 6 leaves the library

Optional: what the interpreter actually does

What you learned

Was this article helpful?