The Edge Cases You've Been Avoiding

Warm-up: from Day 6

The last warm-up of the course, from memory first.

1. Predict which keys survive.

from collections import OrderedDict

cache = OrderedDict()
for key in "abc":
    cache[key] = key.upper()

cache.move_to_end("a")
cache["d"] = "D"
while len(cache) > 3:
    cache.popitem(last=False)

print(list(cache))

['c', 'a', 'd']

2. Fix. Four lookups for three distinct ids should cost one query: this should print 3 1.

queries = 0

def fetch(ids):
    global queries
    queries += 1
    return {i: f"row {i}" for i in ids}

rows = {}
for i in [3, 1, 3, 2]:
    rows.update(fetch([i]))

print(len(rows), queries)

queries = 0

def fetch(ids):
    global queries
    queries += 1
    return {i: f"row {i}" for i in ids}

rows = fetch(list(dict.fromkeys([3, 1, 3, 2])))

print(len(rows), queries)

3 1

3. Explain. No code: why do this library’s tests count queries instead of timing things?

Today

Six days in, the library can create, read and update. It cannot delete, and it has no answer to “someone else changed this row”. Both gaps are small to fill and interesting to design, because both interact with the identity map.

delete: three decisions in eight lines

    def delete(self, connection: Connection) -> Self:
        """Remove this row. The object stays usable, and saving it inserts anew."""
        if self.id is None:
            raise ORMError(f"this {type(self).__name__} has never been saved")
        with transaction(connection):
            connection.execute(f"delete from {self.table} where id = ?", (self.id,))
        type(self)._cache_for(connection).pop(self.id, None)   # or get() would still find it
        self.id = None
        return self

Deleting something unsaved is an error, not a no-op. A caller who deletes an object that was never saved has a bug — a lost save(), a wrong variable — and silence would hide it. The alternative view (“delete is idempotent, doing nothing is fine”) is defensible for a delete_by_id(1) API; for a method on an object that knows whether it exists, raising is better.

The cache must forget it. Without the pop, Post.get(connection, 1) returns the deleted object from memory, cheerfully, with no query. That’s this day’s bug hunt.

self.id = None makes the object reusable. After a delete it looks exactly like a never-saved object, so save() inserts it again with a new id. That’s a choice: the alternative is to mark it dead and raise on any further use. This library picks reusable because it matches how the rest of the API treats id — None means “not in the database” everywhere else, and consistency is worth more here than protecting against an unlikely mistake.

refresh: the answer to a stale cache

The identity map has always had an obvious hole: it hands back an object that was loaded at some point in the past, and rows change. Another process, a migration, raw SQL in a shell — the library can’t know.

    def refresh(self, connection: Connection) -> Self:
        if self.id is None:
            raise ORMError(f"this {type(self).__name__} has never been saved")
        row = connection.execute(
            f"select * from {self.table} where id = ?", (self.id,)
        ).fetchone()
        if row is None:
            raise NotFound(f"no {type(self).__name__} with id {self.id}")
        fields = type(self).fields
        # Validate the whole incoming row before changing anything.
        validated = {name: fields[name].check(row[name]) for name in fields}
        for name, value in validated.items():
            setattr(self, name, value)
        self._remember(connection)
        return self

Two decisions carry this method.

Go through the fields. The values are validated exactly as if a caller had assigned them — Day 3’s lesson, applied to the one method whose whole job is to write attributes from database rows.

Validate the whole row before assigning any of it. If you setattr field by field, a value the fields reject partway through leaves the object half-updated: the new title in place, the old views still there, and an exception on its way up. That breaks the rule Day 4.3 pins down — an operation that raises should leave its subject untouched. Building the validated dict first means the bad value raises before a single attribute has changed. There’s a test for exactly this, and it’s the sort of thing the first draft of this method got wrong.

import sqlite3

connection = sqlite3.connect(":memory:", isolation_level=None)
connection.row_factory = sqlite3.Row
connection.execute("create table post (id integer primary key, views integer)")
connection.execute("insert into post (id, views) values (1, 1)")


class Post:
    cache = {}

    def __init__(self, id, views):
        self.id, self.views = id, views

    @classmethod
    def get(cls, id):
        if id not in cls.cache:
            row = connection.execute("select * from post where id = ?", (id,)).fetchone()
            cls.cache[id] = cls(row["id"], row["views"])
        return cls.cache[id]

    def refresh(self):
        row = connection.execute("select * from post where id = ?", (self.id,)).fetchone()
        self.views = row["views"]
        return self


post = Post.get(1)
print("loaded:            ", post.views)

connection.execute("update post set views = 99 where id = 1")     # somebody else
print("after their update:", Post.get(1).views, "(from the cache)")
print("after refresh:     ", post.refresh().views)
print("and the cached one:", Post.get(1).views, "(same object)")

The last line is the identity map paying off: because get returns the same object, refreshing it updates every reference the program holds. A cache that returned copies would leave stale ones scattered around.

The identity map needs a scope

Here is a bug that stayed hidden until someone used two databases. The cache was keyed by row id alone, so get served whatever it had cached for id 1 — from whichever database loaded it first:

a, b = connect(), connect()
Post(title="only in A", views=1).save(a)
Post.get(b, 1)          # returned A's object, though b has no such row

That is the Day 1 cache-key lesson wearing new clothes. Day 1’s bug was a cache keyed by id when it should have included the class; this one is keyed by id when it should include the database. A row’s identity is only meaningful inside the connection it came from, so the cache has to be scoped that way:

    @classmethod
    def _cache_for(cls, connection: Connection) -> "OrderedDict[int, Any]":
        return cls._instances.setdefault(connection, OrderedDict())

_instances becomes {connection: {id: object}}, and get, get_many, _remember and delete all work through _cache_for(connection). Cross- database collisions vanish, and the guarantee tightens honestly: one object per row per connection. It does introduce a lifetime question — a connection’s entry lives until you drop it — so the library adds a per-model Post.forget(connection) and a Model.forget_all(connection) that clears every model. (Model.forget on the base has no cache of its own; call it on a concrete model.)

Writing down what you won’t fix

Not every hole is closed. save and save_all now undo their own id when a write rolls back, but an object saved inside an outer transaction that the caller then rolls back keeps its id and its cache entry — because the cache and the object’s state don’t join the transaction. Fixing that properly means deferring cache and id changes to the outermost commit, which is a real piece of design, not a patch.

The honest move is to scope the claim precisely and give a workaround that actually works. Model.forget_all(connection) alone does not — it clears the cache, but the object you still hold keeps its phantom id, so re-saving it issues a zero-row UPDATE and re-caches something that isn’t there. The supported recovery is: after a rollback you drive yourself, forget the caches and discard the affected objects, reloading existing rows and rebuilding new ones without the rolled-back id.

with contextlib.suppress(RuntimeError):
    with transaction(connection):
        post.save(connection)
        raise RuntimeError("changed my mind")

Model.forget_all(connection)          # 1. invalidate the caches
post = Post(title="outer", views=1)   # 2. rebuild; don't reuse the stale object
post.save(connection)

The same applies to updates and deletes inside a rolled-back outer unit, not only inserts. There’s a passing test for this recovery and an expectedFailure for the underlying limitation:

class KnownLimitations(Base):
    @unittest.expectedFailure
    def test_a_save_inside_an_outer_rollback_stays_cached(self) -> None:
        """Known limitation: object state is not tied to an outer transaction.

        `save`/`save_all` undo their own id on failure, but an object saved
        inside an outer transaction the caller then rolls back keeps its id and
        its cache entry. The supported recovery (forget + rebuild) has its own
        passing test; a full fix defers cache and id changes to the outermost
        commit.
        """

@unittest.expectedFailure inverts the result: the suite is green while the bug exists, and it goes red the day somebody fixes it — which is exactly when you want to be told, so the note can come out of the README and the test can lose its decorator.

import unittest


class NotFound(LookupError):
    pass


class Session:
    """A tiny store with an identity map."""

    def __init__(self):
        self.rows = {1: "first", 2: "second", 3: "third"}
        self.cache = {}
        self.queries = 0

    def get(self, id):
        if id in self.cache:
            return self.cache[id]
        self.queries += 1
        if id not in self.rows:
            raise NotFound(f"no row {id}")
        self.cache[id] = {"id": id, "value": self.rows[id]}
        return self.cache[id]

    def delete(self, id):
        """TODO: fix. It must
           - raise NotFound for a row that isn't there
           - remove the row
           - leave nothing behind that a later get() could return
           - report how many rows it removed (1), so callers can act on it
        """
        del self.rows[id]


class Deleting(unittest.TestCase):
    def setUp(self):
        self.session = Session()

    def test_the_row_is_gone(self):
        self.session.delete(1)
        with self.assertRaises(NotFound):
            self.session.get(1)

    def test_a_cached_row_is_gone_too(self):
        self.session.get(1)                      # now cached
        self.session.delete(1)
        with self.assertRaises(NotFound):
            self.session.get(1)

    def test_deleting_something_missing_raises(self):
        with self.assertRaises(NotFound):
            self.session.delete(99)

    def test_it_reports_what_it_did(self):
        self.assertEqual(self.session.delete(1), 1)

    def test_other_rows_are_untouched(self):
        self.session.get(2)
        self.session.delete(1)
        self.assertEqual(self.session.get(2)["value"], "second")


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(Deleting)
)

    def delete(self, id):
        if id not in self.rows:
            raise NotFound(f"no row {id}")
        del self.rows[id]
        self.cache.pop(id, None)
        return 1

What you learned

• Every method that touches a row has to decide what it owes the identity map; delete must evict, refresh must write through the fields.
• Validate a whole incoming row before assigning any of it, so a rejected value leaves the object unchanged.
• A cache key must identify the row in its scope: by class (Day 1) and by connection (today), or it hands back the wrong object.
• Choose deliberately between raising and no-op for “that isn’t there”, and make the choice consistent across the API.
• @unittest.expectedFailure turns a known limitation into executable documentation that tells you when it’s fixed — kept few, and scoped honestly.