Last day. Two methods to write, one bug to find, and one document that is
worth more than either.
The checkpoint
Checkpoint
Day 7 checkpoint: delete and refresh
Twenty-four tests: the week’s edge cases (delete, refresh, failed-write state,
cross-database identity, outer-rollback recovery), one regression test per day, and one marked
expectedFailure that you should leave failing. delete and refresh are yours.
models.pyPython"""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.
The identity map is scoped **per connection**: a row's identity only means
something inside the database it came from, so the cache is keyed by
(model, connection, id), not by id alone. See Day 7's edge-cases lesson.
"""
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, ORMError
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]]]
#: Identity maps, one per connection: {connection: {id: object}}, each
#: capped at cache_size and ordered least-recently-used first.
_instances: ClassVar[dict[Connection, "OrderedDict[int, Any]"]]
#: How many loaded objects a model keeps per connection. 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 = {}
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)
# --- the identity map -----------------------------------------------------
@classmethod
def _cache_for(cls, connection: Connection) -> "OrderedDict[int, Any]":
"""This model's identity map for one connection, created on first use."""
return cls._instances.setdefault(connection, OrderedDict())
@classmethod
def forget(cls, connection: Connection) -> None:
"""Drop this model's cache for a connection — call it when you close one,
or after a transaction you rolled back yourself."""
cls._instances.pop(connection, None)
@staticmethod
def forget_all(connection: Connection) -> None:
"""Drop every model's cache for a connection."""
for model in Model.registry.values():
model._instances.pop(connection, None)
def _remember(self, connection: Connection) -> None:
"""Put this object in the identity map for `connection`, evicting the oldest."""
assert self.id is not None
cache = type(self)._cache_for(connection)
cache[self.id] = self
cache.move_to_end(self.id)
while len(cache) > type(self).cache_size:
cache.popitem(last=False) # drop the least recently used
# --- writes ---------------------------------------------------------------
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
was_new = self.id is None
try:
with transaction(connection):
self._write(connection)
except BaseException:
if was_new:
self.id = None # the insert was rolled back; undo the id too
raise
self._remember(connection) # 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)
new = [obj for obj in written if obj.id is None]
try:
with transaction(connection):
for obj in written:
obj._write(connection)
except BaseException:
for obj in new:
obj.id = None # the whole batch was rolled back
raise
for obj in written:
obj._remember(connection)
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]
)
# --- reads ----------------------------------------------------------------
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cache = cls._cache_for(connection)
cached = cache.get(id)
if cached is not None:
cache.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(connection)
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.
"""
cache = cls._cache_for(connection)
wanted = list(dict.fromkeys(ids)) # de-duplicate, keep order
found: dict[int, Self] = {}
missing = []
for id in wanted:
cached = cache.get(id)
if cached is None:
missing.append(id)
else:
cache.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(connection)
found[cast(int, obj.id)] = obj
return {id: found[id] for id in wanted if id in found}
# --- delete and refresh ---------------------------------------------------
def delete(self, connection: Connection) -> Self:
"""Remove this row. The object stays usable, and saving it inserts anew.
TODO: refuse an object that was never saved (ORMError), delete the row
inside a transaction, drop it from this connection's cache so a later
get() cannot still find it, and leave this object ready to save again.
"""
raise NotImplementedError("delete")
def refresh(self, connection: Connection) -> Self:
"""Re-read this row, replacing whatever the object holds.
The identity map can go stale: another process, or raw SQL, can change
a row the library has already loaded. This is the way back.
TODO: refuse an object that was never saved (ORMError), re-read the
row, raise NotFound if it has gone, and put the values back -- validate
the WHOLE row before assigning any of it, so a rejected value leaves
the object unchanged.
"""
raise NotImplementedError("refresh")
@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)
"""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
"""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
Day 7's cache is scoped per connection, so the count to watch is the size of
*this connection's* identity map, `Post._cache_for(connection)` — not
`len(Post._instances)`, which counts connections. This is the script Day 6's
memory lesson quotes, adapted to the per-connection cache.
"""
import tracemalloc
from collections.abc import Sequence
from db import connect
from models import Model, Post
from protocols import Connection
ROWS = 5000
def load(connection: Connection, ids: Sequence[int]) -> None:
for id in ids:
Post.get(connection, id)
def measure(connection: Connection, ids: Sequence[int], cache_size: int) -> None:
Post.cache_size = cache_size
Model.forget_all(connection) # clear this connection's identity map
tracemalloc.start()
before = tracemalloc.take_snapshot()
load(connection, ids)
after = tracemalloc.take_snapshot()
tracemalloc.stop()
held = len(Post._cache_for(connection))
grew = sum(stat.size_diff for stat in after.compare_to(before, "lineno"))
print(f"cache_size={cache_size:6}: cache holds {held:5} objects, "
f"memory grew {grew / 1024:8.0f} KiB")
for stat in after.compare_to(before, "lineno")[:3]:
print(f" {str(stat.traceback[0]):40} {stat.size_diff / 1024:8.0f} KiB")
def main() -> None:
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()
"""The week's edge cases, in one suite.
One test is marked expectedFailure: it describes a limitation the library
knows about and has not fixed. A test is the clearest way to write that down.
"""
import unittest
import transactions
from counting import CountingConnection
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
class Base(unittest.TestCase):
def setUp(self) -> None:
self.real = connect()
self.connection = CountingConnection(self.real)
for model in Model.registry.values():
model._instances.clear()
def titles(self) -> list[str]:
return [row["title"] for row in self.real.execute("select title from post order by id")]
class Deleting(Base):
def test_delete_removes_the_row(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
self.assertEqual(self.titles(), [])
def test_delete_forgets_the_object(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_a_deleted_object_can_be_saved_again(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
post.save(self.connection)
self.assertEqual(self.titles(), ["one"])
self.assertIsNotNone(post.id)
def test_deleting_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).delete(self.connection)
def test_delete_is_a_transaction(self) -> None:
Post.save_all(self.connection, [Post(title="one", views=1),
Post(title="two", views=2)])
Post.get(self.connection, 1).delete(self.connection)
self.assertEqual(transactions._depth, {})
self.assertEqual(self.titles(), ["two"])
class Refreshing(Base):
def test_refresh_picks_up_a_change_made_elsewhere(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 99 where id = ?", (post.id,))
self.assertEqual(post.views, 1)
self.assertEqual(post.refresh(self.connection).views, 99)
def test_refresh_validates_what_it_reads(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 'lots' where id = ?", (post.id,))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
def test_a_rejected_refresh_leaves_the_object_unchanged(self) -> None:
"""The whole row is validated before any attribute is assigned."""
post = Post(title="before", views=1).save(self.connection)
self.real.execute("update post set title = ?, views = ? where id = ?",
("after", "invalid", post.id))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
self.assertEqual((post.title, post.views), ("before", 1),
"a rejected refresh must not partly update the object")
def test_refreshing_a_deleted_row_says_so(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("delete from post")
with self.assertRaises(NotFound):
post.refresh(self.connection)
def test_refreshing_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).refresh(self.connection)
class FailedWritesLeaveCleanState(Base):
"""A rolled-back write must not leave an id on an object with no row."""
def test_a_failed_batch_resets_new_ids(self) -> None:
Post(title="taken", views=1).save(self.connection)
fresh = Post(title="fresh", views=1)
duplicate = Post(title="taken", views=2) # fails, rolling back the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [fresh, duplicate])
self.assertIsNone(fresh.id, "a new object keeps no id after a rolled-back batch")
self.assertIsNone(duplicate.id)
self.assertEqual(self.titles(), ["taken"])
def test_a_new_object_can_be_retried_after_a_failed_batch(self) -> None:
first = Post(title="dup", views=1)
second = Post(title="dup", views=2) # duplicate within the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [first, second])
first.save(self.connection) # retry the good one alone
self.assertEqual(self.titles(), ["dup"])
assert first.id is not None
self.assertEqual(Post.get(self.connection, first.id).views, 1)
def test_a_save_that_fails_on_commit_resets_the_id(self) -> None:
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table node (id integer primary key, "
"parent integer references node(id) deferrable initially deferred)")
class Node(Model, table="node"):
from fields import IntegerField as _Int
parent = _Int(null=True)
self.addCleanup(Model.registry.pop, "node", None)
orphan = Node(parent=999) # references a row that never exists
with self.assertRaises(IntegrityError):
orphan.save(self.connection) # the INSERT is fine; COMMIT fails
self.assertIsNone(orphan.id, "a commit failure must undo the id too")
class CrossDatabase(unittest.TestCase):
def setUp(self) -> None:
for model in Model.registry.values():
model._instances.clear()
def test_two_databases_do_not_share_cached_objects(self) -> None:
a, b = connect(), connect()
Post(title="only in A", views=1).save(a)
with self.assertRaises(NotFound):
Post.get(b, 1) # b has no such row
def test_each_database_keeps_its_own_identity(self) -> None:
a, b = connect(), connect()
Post(title="from a", views=1).save(a)
Post(title="from b", views=2).save(b)
self.assertEqual(Post.get(a, 1).title, "from a")
self.assertEqual(Post.get(b, 1).title, "from b")
def test_forget_all_clears_a_connections_cache(self) -> None:
a = connect()
Post(title="a", views=1).save(a)
self.assertIn(a, Post._instances)
Model.forget_all(a)
self.assertNotIn(a, Post._instances)
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` and `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, because the cache does not join the
transaction. The documented workaround is Model.forget_all(connection)
after a rollback you drive yourself. Fixing it in the library means
deferring cache and id changes to the outermost commit.
"""
post = Post(title="doomed", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
post.save(self.connection)
raise RuntimeError("the caller changed their mind")
self.assertEqual(self.titles(), [])
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_the_documented_outer_rollback_recovery(self) -> None:
"""The supported recovery from the limitation above: forget the caches
AND discard the stale objects, rebuilding fresh ones without the
rolled-back id. `forget_all` alone is not enough — it invalidates the
cache but does not repair an object that kept a phantom id."""
doomed = Post(title="outer", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
doomed.save(self.connection)
raise RuntimeError("abort the outer unit")
Model.forget_all(self.connection) # 1. drop the connection's caches
fresh = Post(title="outer", views=1) # 2. rebuild; don't reuse `doomed`
fresh.save(self.connection)
self.assertEqual(self.titles(), ["outer"])
assert fresh.id is not None
self.assertEqual(Post.get(self.connection, fresh.id).title, "outer")
def test_get_many_does_not_chunk(self) -> None:
"""Known limitation: one query however many ids, so a big enough call
hits the driver's parameter limit.
The limit is a property of the build (sqlite3 reports it through
getlimit; it is 32766 here, and was 999 before SQLite 3.32), so this
test lowers it rather than trying to find the real one.
"""
import sqlite3
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.real.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 10)
self.addCleanup(self.real.setlimit, sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 32766)
with self.assertRaises(sqlite3.OperationalError):
Post.get_many(self.connection, list(range(1, 20)))
class TheWeek(Base):
"""One test per day, so the whole library stays honest."""
def test_day1_two_models_with_the_same_id_stay_apart(self) -> None:
Post(title="one", views=1).save(self.connection)
User(email="someone@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
def test_day2_fields_validate_and_are_per_instance(self) -> None:
first = Post(title="first", views=1)
second = Post(title="second", views=2)
self.assertEqual((first.title, second.title), ("first", "second"))
with self.assertRaises(ValidationError):
first.views = -1
def test_day3_values_from_the_database_are_validated(self) -> None:
self.real.execute("insert into post (title, views) values ('x', 'lots')")
with self.assertRaises(ValidationError):
Post.get(self.connection, 1)
def test_day4_save_all_is_all_or_nothing(self) -> None:
Post(title="one", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="two", views=2),
Post(title="one", views=3)])
self.assertEqual(self.titles(), ["one"])
def test_day6_get_many_is_one_query_and_the_cache_is_bounded(self) -> None:
self.addCleanup(setattr, Post, "cache_size", Post.cache_size)
Post.cache_size = 5
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.connection.reset()
found = Post.get_many(self.connection, list(range(1, 21)))
self.assertEqual(len(found), 20)
self.assertEqual(self.connection.selects, 1)
self.assertLessEqual(len(Post._cache_for(self.connection)), 5)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
# tinyorm
A deliberately small ORM: typed fields, a model registry, get/save/delete and
transactions, over a DB-API-style SQLite connection. No joins, no relationships,
no migrations, no async — on purpose.
Built over seven days in "Designing Reliable Libraries".
## Quickstart
```python
from db import connect
from models import Model
from fields import IntegerField, TextField
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
connection = connect("posts.db")
try:
post = Post(title="Hello", views=0).save(connection)
post.views = 10
post.save(connection)
same = Post.get(connection, post.id) # same object, from the identity map
many = Post.get_many(connection, [1, 2, 3]) # one query
post.delete(connection)
finally:
# The identity map holds a strong reference per connection, so drop it
# before you close the connection -- on the failure path too, which is why
# this is a finally and not just the last two lines of the happy path.
Model.forget_all(connection)
connection.close()
```
## What it guarantees
- **Validation on the way in and the way out.** Every value passes through its
field, whether it came from a caller or from the database. Assigning an
invalid value raises `ValidationError` and changes nothing.
- **Atomic writes.** `save`, `save_all` and `delete` run in a transaction —
including a failure *during* commit — and nested transactions become
savepoints. Note the scope: it is the *database* rows that are all-or-nothing.
A new object's `id` is reset when its own write rolls back, but object state is
not otherwise tied to a transaction (see the limitations).
- **One object per row, per connection, while it is cached.** `get` returns the
same object for the same row *on the same connection*.
- **A domain-error hierarchy.** Everything the library raises for a *domain*
problem inherits from `ORMError`, and each also inherits the built-in it
replaces (`ValidationError` is a `ValueError`, `NotFound` is a `LookupError`).
Programming errors are left as themselves — see below.
## What counts as which error
| Situation | Raises |
| --- | --- |
| A value a field rejects | `ValidationError` (an `ORMError` and a `ValueError`) |
| `get`/`refresh` of a row that isn't there | `NotFound` (an `ORMError` and a `LookupError`) |
| A registered driver integrity error (unique, FK) | `IntegrityError` (an `ORMError`) |
| Reading a field that was never set | `AttributeError` — a programming error, left as-is |
| An unknown column in a constructor | `TypeError` — a programming error, left as-is |
| A driver *operational* error (disk, syntax) | the driver's own exception, uncaught |
Only domain errors are translated. Programming and infrastructure errors are
deliberately allowed through unchanged, so a bug in your code still looks like a
bug in your code.
## What it does not do
| Limitation | Why | Workaround |
| --- | --- | --- |
| Only tested with SQLite | The `Connection` protocol captures the *shape* (`execute`, `commit`, a `?`-placeholder dialect, `sqlite3.Row`-style rows, a read-only `lastrowid`), not a full DB-API | Use the supplied `db.connect`, or an adapter that matches that shape and `execute("?", ...)` placeholders |
| No joins or relationships | Out of scope for the course project | Write SQL and build models from the rows |
| No migrations | Same | `db.SCHEMA` and your own scripts |
| `get_many` never chunks | One query, however many ids | Chunk at the call site; read the limit from `connection.getlimit(SQLITE_LIMIT_VARIABLE_NUMBER)` (32766 on SQLite 3.32+, 999 before) |
| Identity map capped at `cache_size` (128) per connection | An unbounded cache is a leak — measured at 1.7 MB per 5000 rows | Raise a model's `cache_size`, or call `object.refresh(connection)` |
| Object state isn't tied to an *outer* transaction | `save`/`save_all` undo their own id on failure, but a save inside a transaction the caller rolls back keeps its id and cache entry | After such a rollback: `Model.forget_all(connection)` **and** discard the affected objects — reload existing rows and rebuild new objects fresh. `forget_all` invalidates the cache; it does not repair a phantom id on an object you kept. Applies to inserts, updates and deletes |
| Cached values can go stale | Another writer can change a row the cache holds | `object.refresh(connection)` |
| A connection's cache is never dropped automatically | It's keyed by the connection object | A model's `forget(connection)` (e.g. `Post.forget(connection)`), or `Model.forget_all(connection)` for every model, when you close one. `Model.forget(...)` on the base class has no cache of its own — call it on a concrete model |
| One driver per process | `errors.use_driver` is module-level state | Use one database driver, or fork the module |
| Not thread-safe | No locking; the registry and per-connection caches are shared | One connection per thread, and don't share a model's cache across threads |
## Running the tests and the benchmark
```bash
python -m unittest # the suite; one test is an expected failure
python benchmark.py # N+1 versus batched (numbers are illustrative)
python leak.py # the identity map, bounded versus not
mypy --strict --enable-error-code=deprecated .
```
The expected failure is deliberate: it documents the outer-rollback object-state
limitation as executable text rather than prose, and turns red the day it's
fixed.
Hint 1: the idea
Both methods start the same way: an object that was never saved has no row, so
refuse it. After that, each has one rule that’s easy to miss. delete must
also forget the object in this connection’s cache, or get finds it again.
refresh must validate the whole row before assigning any field, so one bad
value leaves the object exactly as it was.
Hint 2: a first step
Write delete first, as in The Edge Cases You’ve Been Avoiding: delete inside
transaction(connection), drop the object from type(self)._cache_for(connection),
and set self.id back to None. For refresh, build a dictionary of every
field’s checked value before the first setattr.
Show one solution
def delete(self, connection: Connection) -> Self: 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 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 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 orderings matter, and both are the same idea: change the database first,
then the object.
In delete, the cache is cleared after the transaction commits. Clear it
first and a failed delete leaves a row in the database that the library has
forgotten — the mirror image of Day 4’s rule about caching after the commit.
In refresh, nothing on the object changes until everything is safe: the
NotFound check comes first, then the whole row is validated into a dict, and
only then is anything assigned. A row that vanished, or a value the fields
reject, leaves the object exactly as it was. That’s the rule worth carrying
elsewhere: an operation that raises should leave its subject untouched.
fields[name].check(...) and setattr, rather than writing _name directly,
is the Day 3 lesson again: a row read back from the database is input, and input
goes through the fields.
Bug hunt: the row that wouldn’t die
One line removed from delete.
ReproducePython"""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.
The identity map is scoped **per connection**: a row's identity only means
something inside the database it came from, so the cache is keyed by
(model, connection, id), not by id alone. See Day 7's edge-cases lesson.
"""
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, ORMError
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]]]
#: Identity maps, one per connection: {connection: {id: object}}, each
#: capped at cache_size and ordered least-recently-used first.
_instances: ClassVar[dict[Connection, "OrderedDict[int, Any]"]]
#: How many loaded objects a model keeps per connection. 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 = {}
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)
# --- the identity map -----------------------------------------------------
@classmethod
def _cache_for(cls, connection: Connection) -> "OrderedDict[int, Any]":
"""This model's identity map for one connection, created on first use."""
return cls._instances.setdefault(connection, OrderedDict())
@classmethod
def forget(cls, connection: Connection) -> None:
"""Drop this model's cache for a connection — call it when you close one,
or after a transaction you rolled back yourself."""
cls._instances.pop(connection, None)
@staticmethod
def forget_all(connection: Connection) -> None:
"""Drop every model's cache for a connection."""
for model in Model.registry.values():
model._instances.pop(connection, None)
def _remember(self, connection: Connection) -> None:
"""Put this object in the identity map for `connection`, evicting the oldest."""
assert self.id is not None
cache = type(self)._cache_for(connection)
cache[self.id] = self
cache.move_to_end(self.id)
while len(cache) > type(self).cache_size:
cache.popitem(last=False) # drop the least recently used
# --- writes ---------------------------------------------------------------
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
was_new = self.id is None
try:
with transaction(connection):
self._write(connection)
except BaseException:
if was_new:
self.id = None # the insert was rolled back; undo the id too
raise
self._remember(connection) # 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)
new = [obj for obj in written if obj.id is None]
try:
with transaction(connection):
for obj in written:
obj._write(connection)
except BaseException:
for obj in new:
obj.id = None # the whole batch was rolled back
raise
for obj in written:
obj._remember(connection)
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]
)
# --- reads ----------------------------------------------------------------
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cache = cls._cache_for(connection)
cached = cache.get(id)
if cached is not None:
cache.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(connection)
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.
"""
cache = cls._cache_for(connection)
wanted = list(dict.fromkeys(ids)) # de-duplicate, keep order
found: dict[int, Self] = {}
missing = []
for id in wanted:
cached = cache.get(id)
if cached is None:
missing.append(id)
else:
cache.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(connection)
found[cast(int, obj.id)] = obj
return {id: found[id] for id in wanted if id in found}
# --- delete and refresh ---------------------------------------------------
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,))
self.id = None
return self
def refresh(self, connection: Connection) -> Self:
"""Re-read this row, replacing whatever the object holds.
The identity map can go stale: another process, or raw SQL, can change
a row the library has already loaded. This is the way back.
"""
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, so a value
# the fields reject leaves this object exactly as it was.
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
@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)
"""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
"""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
Day 7's cache is scoped per connection, so the count to watch is the size of
*this connection's* identity map, `Post._cache_for(connection)` — not
`len(Post._instances)`, which counts connections. This is the script Day 6's
memory lesson quotes, adapted to the per-connection cache.
"""
import tracemalloc
from collections.abc import Sequence
from db import connect
from models import Model, Post
from protocols import Connection
ROWS = 5000
def load(connection: Connection, ids: Sequence[int]) -> None:
for id in ids:
Post.get(connection, id)
def measure(connection: Connection, ids: Sequence[int], cache_size: int) -> None:
Post.cache_size = cache_size
Model.forget_all(connection) # clear this connection's identity map
tracemalloc.start()
before = tracemalloc.take_snapshot()
load(connection, ids)
after = tracemalloc.take_snapshot()
tracemalloc.stop()
held = len(Post._cache_for(connection))
grew = sum(stat.size_diff for stat in after.compare_to(before, "lineno"))
print(f"cache_size={cache_size:6}: cache holds {held:5} objects, "
f"memory grew {grew / 1024:8.0f} KiB")
for stat in after.compare_to(before, "lineno")[:3]:
print(f" {str(stat.traceback[0]):40} {stat.size_diff / 1024:8.0f} KiB")
def main() -> None:
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()
"""The week's edge cases, in one suite.
One test is marked expectedFailure: it describes a limitation the library
knows about and has not fixed. A test is the clearest way to write that down.
"""
import unittest
import transactions
from counting import CountingConnection
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
class Base(unittest.TestCase):
def setUp(self) -> None:
self.real = connect()
self.connection = CountingConnection(self.real)
for model in Model.registry.values():
model._instances.clear()
def titles(self) -> list[str]:
return [row["title"] for row in self.real.execute("select title from post order by id")]
class Deleting(Base):
def test_delete_removes_the_row(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
self.assertEqual(self.titles(), [])
def test_delete_forgets_the_object(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_a_deleted_object_can_be_saved_again(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
post.save(self.connection)
self.assertEqual(self.titles(), ["one"])
self.assertIsNotNone(post.id)
def test_deleting_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).delete(self.connection)
def test_delete_is_a_transaction(self) -> None:
Post.save_all(self.connection, [Post(title="one", views=1),
Post(title="two", views=2)])
Post.get(self.connection, 1).delete(self.connection)
self.assertEqual(transactions._depth, {})
self.assertEqual(self.titles(), ["two"])
class Refreshing(Base):
def test_refresh_picks_up_a_change_made_elsewhere(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 99 where id = ?", (post.id,))
self.assertEqual(post.views, 1)
self.assertEqual(post.refresh(self.connection).views, 99)
def test_refresh_validates_what_it_reads(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 'lots' where id = ?", (post.id,))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
def test_a_rejected_refresh_leaves_the_object_unchanged(self) -> None:
"""The whole row is validated before any attribute is assigned."""
post = Post(title="before", views=1).save(self.connection)
self.real.execute("update post set title = ?, views = ? where id = ?",
("after", "invalid", post.id))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
self.assertEqual((post.title, post.views), ("before", 1),
"a rejected refresh must not partly update the object")
def test_refreshing_a_deleted_row_says_so(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("delete from post")
with self.assertRaises(NotFound):
post.refresh(self.connection)
def test_refreshing_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).refresh(self.connection)
class FailedWritesLeaveCleanState(Base):
"""A rolled-back write must not leave an id on an object with no row."""
def test_a_failed_batch_resets_new_ids(self) -> None:
Post(title="taken", views=1).save(self.connection)
fresh = Post(title="fresh", views=1)
duplicate = Post(title="taken", views=2) # fails, rolling back the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [fresh, duplicate])
self.assertIsNone(fresh.id, "a new object keeps no id after a rolled-back batch")
self.assertIsNone(duplicate.id)
self.assertEqual(self.titles(), ["taken"])
def test_a_new_object_can_be_retried_after_a_failed_batch(self) -> None:
first = Post(title="dup", views=1)
second = Post(title="dup", views=2) # duplicate within the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [first, second])
first.save(self.connection) # retry the good one alone
self.assertEqual(self.titles(), ["dup"])
assert first.id is not None
self.assertEqual(Post.get(self.connection, first.id).views, 1)
def test_a_save_that_fails_on_commit_resets_the_id(self) -> None:
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table node (id integer primary key, "
"parent integer references node(id) deferrable initially deferred)")
class Node(Model, table="node"):
from fields import IntegerField as _Int
parent = _Int(null=True)
self.addCleanup(Model.registry.pop, "node", None)
orphan = Node(parent=999) # references a row that never exists
with self.assertRaises(IntegrityError):
orphan.save(self.connection) # the INSERT is fine; COMMIT fails
self.assertIsNone(orphan.id, "a commit failure must undo the id too")
class CrossDatabase(unittest.TestCase):
def setUp(self) -> None:
for model in Model.registry.values():
model._instances.clear()
def test_two_databases_do_not_share_cached_objects(self) -> None:
a, b = connect(), connect()
Post(title="only in A", views=1).save(a)
with self.assertRaises(NotFound):
Post.get(b, 1) # b has no such row
def test_each_database_keeps_its_own_identity(self) -> None:
a, b = connect(), connect()
Post(title="from a", views=1).save(a)
Post(title="from b", views=2).save(b)
self.assertEqual(Post.get(a, 1).title, "from a")
self.assertEqual(Post.get(b, 1).title, "from b")
def test_forget_all_clears_a_connections_cache(self) -> None:
a = connect()
Post(title="a", views=1).save(a)
self.assertIn(a, Post._instances)
Model.forget_all(a)
self.assertNotIn(a, Post._instances)
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` and `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, because the cache does not join the
transaction. The documented workaround is Model.forget_all(connection)
after a rollback you drive yourself. Fixing it in the library means
deferring cache and id changes to the outermost commit.
"""
post = Post(title="doomed", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
post.save(self.connection)
raise RuntimeError("the caller changed their mind")
self.assertEqual(self.titles(), [])
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_the_documented_outer_rollback_recovery(self) -> None:
"""The supported recovery from the limitation above: forget the caches
AND discard the stale objects, rebuilding fresh ones without the
rolled-back id. `forget_all` alone is not enough — it invalidates the
cache but does not repair an object that kept a phantom id."""
doomed = Post(title="outer", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
doomed.save(self.connection)
raise RuntimeError("abort the outer unit")
Model.forget_all(self.connection) # 1. drop the connection's caches
fresh = Post(title="outer", views=1) # 2. rebuild; don't reuse `doomed`
fresh.save(self.connection)
self.assertEqual(self.titles(), ["outer"])
assert fresh.id is not None
self.assertEqual(Post.get(self.connection, fresh.id).title, "outer")
def test_get_many_does_not_chunk(self) -> None:
"""Known limitation: one query however many ids, so a big enough call
hits the driver's parameter limit.
The limit is a property of the build (sqlite3 reports it through
getlimit; it is 32766 here, and was 999 before SQLite 3.32), so this
test lowers it rather than trying to find the real one.
"""
import sqlite3
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.real.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 10)
self.addCleanup(self.real.setlimit, sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 32766)
with self.assertRaises(sqlite3.OperationalError):
Post.get_many(self.connection, list(range(1, 20)))
class TheWeek(Base):
"""One test per day, so the whole library stays honest."""
def test_day1_two_models_with_the_same_id_stay_apart(self) -> None:
Post(title="one", views=1).save(self.connection)
User(email="someone@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
def test_day2_fields_validate_and_are_per_instance(self) -> None:
first = Post(title="first", views=1)
second = Post(title="second", views=2)
self.assertEqual((first.title, second.title), ("first", "second"))
with self.assertRaises(ValidationError):
first.views = -1
def test_day3_values_from_the_database_are_validated(self) -> None:
self.real.execute("insert into post (title, views) values ('x', 'lots')")
with self.assertRaises(ValidationError):
Post.get(self.connection, 1)
def test_day4_save_all_is_all_or_nothing(self) -> None:
Post(title="one", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="two", views=2),
Post(title="one", views=3)])
self.assertEqual(self.titles(), ["one"])
def test_day6_get_many_is_one_query_and_the_cache_is_bounded(self) -> None:
self.addCleanup(setattr, Post, "cache_size", Post.cache_size)
Post.cache_size = 5
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.connection.reset()
found = Post.get_many(self.connection, list(range(1, 21)))
self.assertEqual(len(found), 20)
self.assertEqual(self.connection.selects, 1)
self.assertLessEqual(len(Post._cache_for(self.connection)), 5)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
# tinyorm
A deliberately small ORM: typed fields, a model registry, get/save/delete and
transactions, over a DB-API-style SQLite connection. No joins, no relationships,
no migrations, no async — on purpose.
Built over seven days in "Designing Reliable Libraries".
## Quickstart
```python
from db import connect
from models import Model
from fields import IntegerField, TextField
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
connection = connect("posts.db")
try:
post = Post(title="Hello", views=0).save(connection)
post.views = 10
post.save(connection)
same = Post.get(connection, post.id) # same object, from the identity map
many = Post.get_many(connection, [1, 2, 3]) # one query
post.delete(connection)
finally:
# The identity map holds a strong reference per connection, so drop it
# before you close the connection -- on the failure path too, which is why
# this is a finally and not just the last two lines of the happy path.
Model.forget_all(connection)
connection.close()
```
## What it guarantees
- **Validation on the way in and the way out.** Every value passes through its
field, whether it came from a caller or from the database. Assigning an
invalid value raises `ValidationError` and changes nothing.
- **Atomic writes.** `save`, `save_all` and `delete` run in a transaction —
including a failure *during* commit — and nested transactions become
savepoints. Note the scope: it is the *database* rows that are all-or-nothing.
A new object's `id` is reset when its own write rolls back, but object state is
not otherwise tied to a transaction (see the limitations).
- **One object per row, per connection, while it is cached.** `get` returns the
same object for the same row *on the same connection*.
- **A domain-error hierarchy.** Everything the library raises for a *domain*
problem inherits from `ORMError`, and each also inherits the built-in it
replaces (`ValidationError` is a `ValueError`, `NotFound` is a `LookupError`).
Programming errors are left as themselves — see below.
## What counts as which error
| Situation | Raises |
| --- | --- |
| A value a field rejects | `ValidationError` (an `ORMError` and a `ValueError`) |
| `get`/`refresh` of a row that isn't there | `NotFound` (an `ORMError` and a `LookupError`) |
| A registered driver integrity error (unique, FK) | `IntegrityError` (an `ORMError`) |
| Reading a field that was never set | `AttributeError` — a programming error, left as-is |
| An unknown column in a constructor | `TypeError` — a programming error, left as-is |
| A driver *operational* error (disk, syntax) | the driver's own exception, uncaught |
Only domain errors are translated. Programming and infrastructure errors are
deliberately allowed through unchanged, so a bug in your code still looks like a
bug in your code.
## What it does not do
| Limitation | Why | Workaround |
| --- | --- | --- |
| Only tested with SQLite | The `Connection` protocol captures the *shape* (`execute`, `commit`, a `?`-placeholder dialect, `sqlite3.Row`-style rows, a read-only `lastrowid`), not a full DB-API | Use the supplied `db.connect`, or an adapter that matches that shape and `execute("?", ...)` placeholders |
| No joins or relationships | Out of scope for the course project | Write SQL and build models from the rows |
| No migrations | Same | `db.SCHEMA` and your own scripts |
| `get_many` never chunks | One query, however many ids | Chunk at the call site; read the limit from `connection.getlimit(SQLITE_LIMIT_VARIABLE_NUMBER)` (32766 on SQLite 3.32+, 999 before) |
| Identity map capped at `cache_size` (128) per connection | An unbounded cache is a leak — measured at 1.7 MB per 5000 rows | Raise a model's `cache_size`, or call `object.refresh(connection)` |
| Object state isn't tied to an *outer* transaction | `save`/`save_all` undo their own id on failure, but a save inside a transaction the caller rolls back keeps its id and cache entry | After such a rollback: `Model.forget_all(connection)` **and** discard the affected objects — reload existing rows and rebuild new objects fresh. `forget_all` invalidates the cache; it does not repair a phantom id on an object you kept. Applies to inserts, updates and deletes |
| Cached values can go stale | Another writer can change a row the cache holds | `object.refresh(connection)` |
| A connection's cache is never dropped automatically | It's keyed by the connection object | A model's `forget(connection)` (e.g. `Post.forget(connection)`), or `Model.forget_all(connection)` for every model, when you close one. `Model.forget(...)` on the base class has no cache of its own — call it on a concrete model |
| One driver per process | `errors.use_driver` is module-level state | Use one database driver, or fork the module |
| Not thread-safe | No locking; the registry and per-connection caches are shared | One connection per thread, and don't share a model's cache across threads |
## Running the tests and the benchmark
```bash
python -m unittest # the suite; one test is an expected failure
python benchmark.py # N+1 versus batched (numbers are illustrative)
python leak.py # the identity map, bounded versus not
mypy --strict --enable-error-code=deprecated .
```
The expected failure is deliberate: it documents the outer-rollback object-state
limitation as executable text rather than prose, and turns red the day it's
fixed.
Reproduce. One failure, and it isn’t the obvious one:
FAIL: test_delete_forgets_the_objectAssertionError: NotFound not raised
test_delete_removes_the_row passes — the row really is gone from the
database. Only the test that reads it back afterwards fails.
Inspect.Post.get(connection, 1) returns a Post. The row does not
exist. Add a counter and you’ll see the query was never issued.
Narrow.delete removed the row but left the object in the connection’s
identity map. get checks that map first, finds it, and returns it — the cache
doing exactly what it was built to do, on an object that should no longer be in
it.
Pin.test_delete_forgets_the_object is the regression test. Note how
little it asserts, and how much it covers: any future change that deletes
without invalidating fails here.
Fix.
type(self)._cache_for(connection).pop(self.id, None) # or get() would still find it
Every cache needs an invalidation story
This is the week’s last word on the identity map: Day 1 introduced it (and its
first bug hunt was a wrong cache key), Day 6 bounded it, and today it had to be
invalidated on delete and scoped per connection. A cache is never one feature —
it is a feature plus an eviction policy plus an invalidation rule plus a scope,
and every one of those but the first is where the bugs live.
The finished library
Day 7, completePython"""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.
The identity map is scoped **per connection**: a row's identity only means
something inside the database it came from, so the cache is keyed by
(model, connection, id), not by id alone. See Day 7's edge-cases lesson.
"""
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, ORMError
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]]]
#: Identity maps, one per connection: {connection: {id: object}}, each
#: capped at cache_size and ordered least-recently-used first.
_instances: ClassVar[dict[Connection, "OrderedDict[int, Any]"]]
#: How many loaded objects a model keeps per connection. 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 = {}
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)
# --- the identity map -----------------------------------------------------
@classmethod
def _cache_for(cls, connection: Connection) -> "OrderedDict[int, Any]":
"""This model's identity map for one connection, created on first use."""
return cls._instances.setdefault(connection, OrderedDict())
@classmethod
def forget(cls, connection: Connection) -> None:
"""Drop this model's cache for a connection — call it when you close one,
or after a transaction you rolled back yourself."""
cls._instances.pop(connection, None)
@staticmethod
def forget_all(connection: Connection) -> None:
"""Drop every model's cache for a connection."""
for model in Model.registry.values():
model._instances.pop(connection, None)
def _remember(self, connection: Connection) -> None:
"""Put this object in the identity map for `connection`, evicting the oldest."""
assert self.id is not None
cache = type(self)._cache_for(connection)
cache[self.id] = self
cache.move_to_end(self.id)
while len(cache) > type(self).cache_size:
cache.popitem(last=False) # drop the least recently used
# --- writes ---------------------------------------------------------------
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
was_new = self.id is None
try:
with transaction(connection):
self._write(connection)
except BaseException:
if was_new:
self.id = None # the insert was rolled back; undo the id too
raise
self._remember(connection) # 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)
new = [obj for obj in written if obj.id is None]
try:
with transaction(connection):
for obj in written:
obj._write(connection)
except BaseException:
for obj in new:
obj.id = None # the whole batch was rolled back
raise
for obj in written:
obj._remember(connection)
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]
)
# --- reads ----------------------------------------------------------------
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cache = cls._cache_for(connection)
cached = cache.get(id)
if cached is not None:
cache.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(connection)
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.
"""
cache = cls._cache_for(connection)
wanted = list(dict.fromkeys(ids)) # de-duplicate, keep order
found: dict[int, Self] = {}
missing = []
for id in wanted:
cached = cache.get(id)
if cached is None:
missing.append(id)
else:
cache.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(connection)
found[cast(int, obj.id)] = obj
return {id: found[id] for id in wanted if id in found}
# --- delete and refresh ---------------------------------------------------
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
def refresh(self, connection: Connection) -> Self:
"""Re-read this row, replacing whatever the object holds.
The identity map can go stale: another process, or raw SQL, can change
a row the library has already loaded. This is the way back.
"""
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, so a value
# the fields reject leaves this object exactly as it was.
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
@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)
"""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
"""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
Day 7's cache is scoped per connection, so the count to watch is the size of
*this connection's* identity map, `Post._cache_for(connection)` — not
`len(Post._instances)`, which counts connections. This is the script Day 6's
memory lesson quotes, adapted to the per-connection cache.
"""
import tracemalloc
from collections.abc import Sequence
from db import connect
from models import Model, Post
from protocols import Connection
ROWS = 5000
def load(connection: Connection, ids: Sequence[int]) -> None:
for id in ids:
Post.get(connection, id)
def measure(connection: Connection, ids: Sequence[int], cache_size: int) -> None:
Post.cache_size = cache_size
Model.forget_all(connection) # clear this connection's identity map
tracemalloc.start()
before = tracemalloc.take_snapshot()
load(connection, ids)
after = tracemalloc.take_snapshot()
tracemalloc.stop()
held = len(Post._cache_for(connection))
grew = sum(stat.size_diff for stat in after.compare_to(before, "lineno"))
print(f"cache_size={cache_size:6}: cache holds {held:5} objects, "
f"memory grew {grew / 1024:8.0f} KiB")
for stat in after.compare_to(before, "lineno")[:3]:
print(f" {str(stat.traceback[0]):40} {stat.size_diff / 1024:8.0f} KiB")
def main() -> None:
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()
"""The week's edge cases, in one suite.
One test is marked expectedFailure: it describes a limitation the library
knows about and has not fixed. A test is the clearest way to write that down.
"""
import unittest
import transactions
from counting import CountingConnection
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
class Base(unittest.TestCase):
def setUp(self) -> None:
self.real = connect()
self.connection = CountingConnection(self.real)
for model in Model.registry.values():
model._instances.clear()
def titles(self) -> list[str]:
return [row["title"] for row in self.real.execute("select title from post order by id")]
class Deleting(Base):
def test_delete_removes_the_row(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
self.assertEqual(self.titles(), [])
def test_delete_forgets_the_object(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_a_deleted_object_can_be_saved_again(self) -> None:
post = Post(title="one", views=1).save(self.connection)
post.delete(self.connection)
post.save(self.connection)
self.assertEqual(self.titles(), ["one"])
self.assertIsNotNone(post.id)
def test_deleting_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).delete(self.connection)
def test_delete_is_a_transaction(self) -> None:
Post.save_all(self.connection, [Post(title="one", views=1),
Post(title="two", views=2)])
Post.get(self.connection, 1).delete(self.connection)
self.assertEqual(transactions._depth, {})
self.assertEqual(self.titles(), ["two"])
class Refreshing(Base):
def test_refresh_picks_up_a_change_made_elsewhere(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 99 where id = ?", (post.id,))
self.assertEqual(post.views, 1)
self.assertEqual(post.refresh(self.connection).views, 99)
def test_refresh_validates_what_it_reads(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("update post set views = 'lots' where id = ?", (post.id,))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
def test_a_rejected_refresh_leaves_the_object_unchanged(self) -> None:
"""The whole row is validated before any attribute is assigned."""
post = Post(title="before", views=1).save(self.connection)
self.real.execute("update post set title = ?, views = ? where id = ?",
("after", "invalid", post.id))
with self.assertRaises(ValidationError):
post.refresh(self.connection)
self.assertEqual((post.title, post.views), ("before", 1),
"a rejected refresh must not partly update the object")
def test_refreshing_a_deleted_row_says_so(self) -> None:
post = Post(title="one", views=1).save(self.connection)
self.real.execute("delete from post")
with self.assertRaises(NotFound):
post.refresh(self.connection)
def test_refreshing_something_unsaved_is_an_error(self) -> None:
with self.assertRaises(ORMError):
Post(title="never saved", views=0).refresh(self.connection)
class FailedWritesLeaveCleanState(Base):
"""A rolled-back write must not leave an id on an object with no row."""
def test_a_failed_batch_resets_new_ids(self) -> None:
Post(title="taken", views=1).save(self.connection)
fresh = Post(title="fresh", views=1)
duplicate = Post(title="taken", views=2) # fails, rolling back the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [fresh, duplicate])
self.assertIsNone(fresh.id, "a new object keeps no id after a rolled-back batch")
self.assertIsNone(duplicate.id)
self.assertEqual(self.titles(), ["taken"])
def test_a_new_object_can_be_retried_after_a_failed_batch(self) -> None:
first = Post(title="dup", views=1)
second = Post(title="dup", views=2) # duplicate within the batch
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [first, second])
first.save(self.connection) # retry the good one alone
self.assertEqual(self.titles(), ["dup"])
assert first.id is not None
self.assertEqual(Post.get(self.connection, first.id).views, 1)
def test_a_save_that_fails_on_commit_resets_the_id(self) -> None:
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table node (id integer primary key, "
"parent integer references node(id) deferrable initially deferred)")
class Node(Model, table="node"):
from fields import IntegerField as _Int
parent = _Int(null=True)
self.addCleanup(Model.registry.pop, "node", None)
orphan = Node(parent=999) # references a row that never exists
with self.assertRaises(IntegrityError):
orphan.save(self.connection) # the INSERT is fine; COMMIT fails
self.assertIsNone(orphan.id, "a commit failure must undo the id too")
class CrossDatabase(unittest.TestCase):
def setUp(self) -> None:
for model in Model.registry.values():
model._instances.clear()
def test_two_databases_do_not_share_cached_objects(self) -> None:
a, b = connect(), connect()
Post(title="only in A", views=1).save(a)
with self.assertRaises(NotFound):
Post.get(b, 1) # b has no such row
def test_each_database_keeps_its_own_identity(self) -> None:
a, b = connect(), connect()
Post(title="from a", views=1).save(a)
Post(title="from b", views=2).save(b)
self.assertEqual(Post.get(a, 1).title, "from a")
self.assertEqual(Post.get(b, 1).title, "from b")
def test_forget_all_clears_a_connections_cache(self) -> None:
a = connect()
Post(title="a", views=1).save(a)
self.assertIn(a, Post._instances)
Model.forget_all(a)
self.assertNotIn(a, Post._instances)
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` and `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, because the cache does not join the
transaction. The documented workaround is Model.forget_all(connection)
after a rollback you drive yourself. Fixing it in the library means
deferring cache and id changes to the outermost commit.
"""
post = Post(title="doomed", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
post.save(self.connection)
raise RuntimeError("the caller changed their mind")
self.assertEqual(self.titles(), [])
with self.assertRaises(NotFound):
Post.get(self.connection, 1)
def test_the_documented_outer_rollback_recovery(self) -> None:
"""The supported recovery from the limitation above: forget the caches
AND discard the stale objects, rebuilding fresh ones without the
rolled-back id. `forget_all` alone is not enough — it invalidates the
cache but does not repair an object that kept a phantom id."""
doomed = Post(title="outer", views=1)
with self.assertRaises(RuntimeError):
with transactions.transaction(self.connection):
doomed.save(self.connection)
raise RuntimeError("abort the outer unit")
Model.forget_all(self.connection) # 1. drop the connection's caches
fresh = Post(title="outer", views=1) # 2. rebuild; don't reuse `doomed`
fresh.save(self.connection)
self.assertEqual(self.titles(), ["outer"])
assert fresh.id is not None
self.assertEqual(Post.get(self.connection, fresh.id).title, "outer")
def test_get_many_does_not_chunk(self) -> None:
"""Known limitation: one query however many ids, so a big enough call
hits the driver's parameter limit.
The limit is a property of the build (sqlite3 reports it through
getlimit; it is 32766 here, and was 999 before SQLite 3.32), so this
test lowers it rather than trying to find the real one.
"""
import sqlite3
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.real.setlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 10)
self.addCleanup(self.real.setlimit, sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER, 32766)
with self.assertRaises(sqlite3.OperationalError):
Post.get_many(self.connection, list(range(1, 20)))
class TheWeek(Base):
"""One test per day, so the whole library stays honest."""
def test_day1_two_models_with_the_same_id_stay_apart(self) -> None:
Post(title="one", views=1).save(self.connection)
User(email="someone@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
def test_day2_fields_validate_and_are_per_instance(self) -> None:
first = Post(title="first", views=1)
second = Post(title="second", views=2)
self.assertEqual((first.title, second.title), ("first", "second"))
with self.assertRaises(ValidationError):
first.views = -1
def test_day3_values_from_the_database_are_validated(self) -> None:
self.real.execute("insert into post (title, views) values ('x', 'lots')")
with self.assertRaises(ValidationError):
Post.get(self.connection, 1)
def test_day4_save_all_is_all_or_nothing(self) -> None:
Post(title="one", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="two", views=2),
Post(title="one", views=3)])
self.assertEqual(self.titles(), ["one"])
def test_day6_get_many_is_one_query_and_the_cache_is_bounded(self) -> None:
self.addCleanup(setattr, Post, "cache_size", Post.cache_size)
Post.cache_size = 5
Post.save_all(self.connection, [Post(title=f"post {index}", views=index)
for index in range(20)])
for model in Model.registry.values():
model._instances.clear()
self.connection.reset()
found = Post.get_many(self.connection, list(range(1, 21)))
self.assertEqual(len(found), 20)
self.assertEqual(self.connection.selects, 1)
self.assertLessEqual(len(Post._cache_for(self.connection)), 5)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
# tinyorm
A deliberately small ORM: typed fields, a model registry, get/save/delete and
transactions, over a DB-API-style SQLite connection. No joins, no relationships,
no migrations, no async — on purpose.
Built over seven days in "Designing Reliable Libraries".
## Quickstart
```python
from db import connect
from models import Model
from fields import IntegerField, TextField
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
connection = connect("posts.db")
try:
post = Post(title="Hello", views=0).save(connection)
post.views = 10
post.save(connection)
same = Post.get(connection, post.id) # same object, from the identity map
many = Post.get_many(connection, [1, 2, 3]) # one query
post.delete(connection)
finally:
# The identity map holds a strong reference per connection, so drop it
# before you close the connection -- on the failure path too, which is why
# this is a finally and not just the last two lines of the happy path.
Model.forget_all(connection)
connection.close()
```
## What it guarantees
- **Validation on the way in and the way out.** Every value passes through its
field, whether it came from a caller or from the database. Assigning an
invalid value raises `ValidationError` and changes nothing.
- **Atomic writes.** `save`, `save_all` and `delete` run in a transaction —
including a failure *during* commit — and nested transactions become
savepoints. Note the scope: it is the *database* rows that are all-or-nothing.
A new object's `id` is reset when its own write rolls back, but object state is
not otherwise tied to a transaction (see the limitations).
- **One object per row, per connection, while it is cached.** `get` returns the
same object for the same row *on the same connection*.
- **A domain-error hierarchy.** Everything the library raises for a *domain*
problem inherits from `ORMError`, and each also inherits the built-in it
replaces (`ValidationError` is a `ValueError`, `NotFound` is a `LookupError`).
Programming errors are left as themselves — see below.
## What counts as which error
| Situation | Raises |
| --- | --- |
| A value a field rejects | `ValidationError` (an `ORMError` and a `ValueError`) |
| `get`/`refresh` of a row that isn't there | `NotFound` (an `ORMError` and a `LookupError`) |
| A registered driver integrity error (unique, FK) | `IntegrityError` (an `ORMError`) |
| Reading a field that was never set | `AttributeError` — a programming error, left as-is |
| An unknown column in a constructor | `TypeError` — a programming error, left as-is |
| A driver *operational* error (disk, syntax) | the driver's own exception, uncaught |
Only domain errors are translated. Programming and infrastructure errors are
deliberately allowed through unchanged, so a bug in your code still looks like a
bug in your code.
## What it does not do
| Limitation | Why | Workaround |
| --- | --- | --- |
| Only tested with SQLite | The `Connection` protocol captures the *shape* (`execute`, `commit`, a `?`-placeholder dialect, `sqlite3.Row`-style rows, a read-only `lastrowid`), not a full DB-API | Use the supplied `db.connect`, or an adapter that matches that shape and `execute("?", ...)` placeholders |
| No joins or relationships | Out of scope for the course project | Write SQL and build models from the rows |
| No migrations | Same | `db.SCHEMA` and your own scripts |
| `get_many` never chunks | One query, however many ids | Chunk at the call site; read the limit from `connection.getlimit(SQLITE_LIMIT_VARIABLE_NUMBER)` (32766 on SQLite 3.32+, 999 before) |
| Identity map capped at `cache_size` (128) per connection | An unbounded cache is a leak — measured at 1.7 MB per 5000 rows | Raise a model's `cache_size`, or call `object.refresh(connection)` |
| Object state isn't tied to an *outer* transaction | `save`/`save_all` undo their own id on failure, but a save inside a transaction the caller rolls back keeps its id and cache entry | After such a rollback: `Model.forget_all(connection)` **and** discard the affected objects — reload existing rows and rebuild new objects fresh. `forget_all` invalidates the cache; it does not repair a phantom id on an object you kept. Applies to inserts, updates and deletes |
| Cached values can go stale | Another writer can change a row the cache holds | `object.refresh(connection)` |
| A connection's cache is never dropped automatically | It's keyed by the connection object | A model's `forget(connection)` (e.g. `Post.forget(connection)`), or `Model.forget_all(connection)` for every model, when you close one. `Model.forget(...)` on the base class has no cache of its own — call it on a concrete model |
| One driver per process | `errors.use_driver` is module-level state | Use one database driver, or fork the module |
| Not thread-safe | No locking; the registry and per-connection caches are shared | One connection per thread, and don't share a model's cache across threads |
## Running the tests and the benchmark
```bash
python -m unittest # the suite; one test is an expected failure
python benchmark.py # N+1 versus batched (numbers are illustrative)
python leak.py # the identity map, bounded versus not
mypy --strict --enable-error-code=deprecated .
```
The expected failure is deliberate: it documents the outer-rollback object-state
limitation as executable text rather than prose, and turns red the day it's
fixed.
Working on your own machine? Download the Day 7 files —
the whole library, its tests, its README and its benchmark:
python -m unittest # 24 tests, one an expected failurepython benchmark.py # N+1 versus batched (numbers are illustrative)python leak.py # the identity map, bounded versus notmypy --strict --enable-error-code=deprecated --python-version 3.14 .
The write-up
This is the real capstone, and it’s writing rather than code. Produce two
things for your library.
A README with the two sections from this morning: what it guarantees, and
what it doesn’t do with a workaround for each. The version in the download is
a model to follow, not to copy — yours should be about what you built.
Three or four decision records, each with Context, Decision and
Consequences. The week hands you the candidates, and for each one you already
know the cost, because you measured it or a bug hunt showed it to you:
Decision
The cost you already know
An identity map
Stale reads, invalidation bugs, memory — three of the week’s bug hunts
Validating rows on load
Roughly a third of the CPU in a load-heavy profile; no way to inspect a bad row through the library
A bounded cache
”Same object for the same row” is now only true for recent rows
The identity map scoped per connection
Correct across databases, but cache entries live until you forget a connection
Errors inheriting built-ins
except ValueError around your library now catches your errors too
errors.use_driver as module state
One driver per process, and test pollution
No chunking in get_many
A big enough call hits the driver’s parameter limit
Write the Consequences section before the Decision section if you can. It’s
harder, and it’s the one a reader needs.
A test can be the documentation
@unittest.expectedFailure on the outer-rollback test says more than a
paragraph would, and it tells you when the limitation is fixed. Where a
limitation is executable, prefer the test.
What the week was about
The library is small and the techniques weren’t the point. Every day had the
same shape underneath:
Day 1 — names refer to objects, and shared state is the default failure
mode.
Day 2 — descriptors and __init_subclass__ let a library enforce its own
invariants instead of documenting them.
Day 3 — types describe what the code believes; only validation makes it
true at a boundary.
Day 4 — an operation that can fail halfway needs a transaction, an
exception hierarchy, and tests for the failure paths.
Day 5 — concurrency is cheap and cleanup is not; cancellation is the
path nobody tests.
Day 6 — measure before you optimise, and measure the condition your users
are in.
Day 7 — write down what you chose and what it cost.
The habit worth keeping is the bug hunt: reproduce, inspect, narrow, pin, fix.
Seven of them this week, and every one was the same underlying mistake in a new
costume — state living somewhere longer-lived than the thing it describes.
Two more pages
The ORM was built to a plan. Two short pages finish the course:
On Your Own — a different library from a
blank file, where the API, the guarantees and the tests are all yours.
Come Back Later — review challenges for
a week from now and a month from now.
What you learned
delete and refresh complete the API, and both are mostly decisions about
the identity map.
Change the database first, then the object; an operation that raises should
leave its subject untouched.
A cache is a feature plus an eviction policy plus an invalidation rule.
The design write-up is the deliverable that outlives the code — and its
Consequences section is the part that matters.
Your progress is saved in this browser.
Sign in
to keep it on every device.