The library now has its own errors, and save() and save_all() are written
against a transaction() that doesn’t exist yet. Fourteen tests describe it.
The checkpoint
Checkpoint
Day 4 checkpoint: all or nothing
transactions.py gives you three helpers — _rollback, _finish and
_translate — and an empty transaction. Write it so the whole suite passes,
including the tests that are about tidiness rather than correctness: the
driver’s error must survive as __cause__, _depth must be empty once the
outermost transaction ends, a failure during commit must roll back too, and a
rollback that fails on its own must not replace the error that caused it.
transactions.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.
"""
from collections.abc import Iterable
from typing import Any, ClassVar, Self, cast
from warnings import deprecated
from errors import NotFound
from fields import Field, IntegerField, TextField
from protocols import Connection
from transactions import transaction
class Model:
#: Every model, by table name.
registry: ClassVar[dict[str, type["Model"]]] = {}
table: ClassVar[str]
fields: ClassVar[dict[str, Field[Any]]]
_instances: ClassVar[dict[int, Any]]
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)
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
with transaction(connection):
self._write(connection)
self._remember() # only once the write is committed
return self
@classmethod
def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None:
"""Write several objects in one transaction: all of them, or none."""
written = list(objects)
with transaction(connection):
for obj in written:
obj._write(connection)
for obj in written:
obj._remember()
def _write(self, connection: Connection) -> None:
columns = self.columns
values = [getattr(self, name) for name in columns]
names = ", ".join(columns)
if self.id is None:
marks = ", ".join("?" for _ in columns)
cursor = connection.execute(
f"insert into {self.table} ({names}) values ({marks})", values
)
self.id = cursor.lastrowid
else:
assignments = ", ".join(f"{name} = ?" for name in columns)
connection.execute(
f"update {self.table} set {assignments} where id = ?", [*values, self.id]
)
def _remember(self) -> None:
"""Put this object in its model's identity map."""
assert self.id is not None
type(self)._instances[self.id] = self
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cached = cls._instances.get(id)
if cached is not None:
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()})
cls._instances[id] = found
return found
@classmethod
@deprecated("Model.fetch is deprecated; use Model.get instead")
def fetch(cls, connection: Connection, id: int) -> Self:
return cls.get(connection, id)
def __repr__(self) -> str:
shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields)
return f"{type(self).__name__}(id={self.id!r}, {shown})"
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
class User(Model, table="user"):
email = TextField(max_length=200)
"""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 — including a failure
during commit itself. Nesting becomes savepoints.
TODO, using the helpers above:
- "begin" at depth 0, else "savepoint orm_<depth>"; record the new depth
- yield the connection, then _finish() it (commit/release) INSIDE the try,
so a failure while committing is handled like any other
- on any exception: _rollback() (guard it — a conflict clause may have
rolled back already, and that secondary error must not mask the first;
add_note() it onto the primary error instead of dropping it), then
_translate() the error, then raise
- in a finally, restore the previous depth (pop it at depth 0)
"""
raise NotImplementedError("transaction")
yield connection # keeps this a generator function
"""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,)
"""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
"""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: ...
@runtime_checkable
class Connection(Protocol):
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: ...
def commit(self) -> None: ...
"""SQLite scaffolding, supplied by the course.
`isolation_level=None` turns off the driver's own implicit transactions, so the
library's `transaction()` is the only thing issuing begin, commit and rollback.
"""
import sqlite3
import errors
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null unique,
views integer not null default 0
);
create table if not exists user (
id integer primary key autoincrement,
email text not null
);
"""
def connect(path: str = ":memory:") -> sqlite3.Connection:
connection = sqlite3.connect(path, isolation_level=None)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
errors.use_driver(integrity=sqlite3.IntegrityError)
return connection
import pathlib
import sqlite3
import tempfile
import unittest
import errors
import transactions
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
from transactions import transaction
class Hierarchy(unittest.TestCase):
def test_every_error_is_an_orm_error(self) -> None:
for error in (ValidationError, NotFound, IntegrityError):
self.assertTrue(issubclass(error, ORMError), error.__name__)
def test_errors_keep_their_builtin_meaning(self) -> None:
self.assertTrue(issubclass(ValidationError, ValueError))
self.assertTrue(issubclass(NotFound, LookupError))
def test_a_field_raises_validation_error(self) -> None:
with self.assertRaises(ValidationError):
Post(title="Hello", views=-1)
def test_old_code_catching_valueerror_still_works(self) -> None:
try:
Post(title="Hello", views=-1)
except ValueError as error:
self.assertIsInstance(error, ValidationError)
else:
self.fail("no error raised")
class Transactions(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def rows(self) -> list[str]:
return [row["title"] for row in self.connection.execute("select title from post")]
def test_save_all_is_all_or_nothing(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="second", views=1),
Post(title="first", views=2)])
self.assertEqual(self.rows(), ["first"], "the good row was rolled back too")
def test_the_driver_error_is_the_cause(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError) as caught:
Post(title="first", views=2).save(self.connection)
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
def test_any_exception_rolls_back(self) -> None:
with self.assertRaises(ZeroDivisionError):
with transaction(self.connection):
Post(title="doomed", views=1)._write(self.connection)
1 / 0
self.assertEqual(self.rows(), [])
def test_a_nested_failure_keeps_the_outer_work(self) -> None:
with transaction(self.connection):
Post(title="outer", views=1)._write(self.connection)
try:
with transaction(self.connection):
Post(title="inner", views=1)._write(self.connection)
raise RuntimeError("no good")
except RuntimeError:
pass
self.assertEqual(self.rows(), ["outer"])
def test_a_failed_save_is_not_cached(self) -> None:
Post(title="first", views=1).save(self.connection)
doomed = Post(title="first", views=2)
with self.assertRaises(IntegrityError):
doomed.save(self.connection)
self.assertNotIn(doomed, Post._instances.values())
def test_nothing_is_left_behind_between_transactions(self) -> None:
Post(title="first", views=1).save(self.connection)
self.assertEqual(transactions._depth, {}, "a connection stayed in the depth map")
def test_a_failure_during_commit_rolls_back_and_recovers(self) -> None:
"""A deferred constraint fails at COMMIT, not at the offending statement."""
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table parent (id integer primary key)")
self.connection.execute(
"create table child (parent_id integer references parent(id) "
"deferrable initially deferred)"
)
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute("insert into child values (999)")
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
# The connection must be usable afterwards, not stuck mid-transaction.
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
with transaction(self.connection):
self.connection.execute("insert into parent values (1)")
self.assertEqual(self.connection.execute("select count(*) from parent").fetchone()[0], 1)
def test_a_conflict_clause_that_self_rolls_back_still_reports_our_error(self) -> None:
"""INSERT OR ROLLBACK ends the transaction itself, so our rollback can't
run — but the caller must still get IntegrityError, not the secondary
'no transaction is active' OperationalError."""
self.connection.execute("insert into post (title, views) values ('same', 0)")
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute(
"insert or rollback into post (title, views) values ('same', 1)")
# the cleanup's own failure is recorded as a note, not raised in place
# of the real error -- and the translation carries the note across
self.assertTrue(any("rollback after this error also failed" in note
for note in getattr(caught.exception, "__notes__", ())))
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
# and the connection is still usable
with transaction(self.connection):
self.connection.execute("insert into post (title, views) values ('other', 2)")
self.assertEqual(sorted(self.rows()), ["other", "same"])
def test_a_commit_really_reaches_the_file(self) -> None:
"""An in-memory database can't tell a commit from an open transaction."""
with tempfile.TemporaryDirectory() as folder:
path = str(pathlib.Path(folder) / "posts.db")
first = connect(path)
Post(title="kept", views=1).save(first)
first.close()
second = connect(path)
titles = [row["title"] for row in second.execute("select title from post")]
second.close()
self.assertEqual(titles, ["kept"])
class StillWorks(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_round_trip(self) -> None:
post = Post(title="Hello", views=3).save(self.connection)
self.assertEqual(Post.get(self.connection, post.id).views, 3)
def test_missing_rows_raise_not_found(self) -> None:
with self.assertRaises(NotFound):
Post.get(self.connection, 99)
def test_two_models_stay_apart(self) -> None:
Post(title="Hello", views=0).save(self.connection)
User(email="a@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Hint 1: the idea
The shape from All or Nothing: begin (or a savepoint, when nested), yield,
finish; on any exception, roll back, translate, re-raise; and in a finally,
restore the depth. The helpers do each step — your job is where each one goes.
One placement matters above the rest: a commit can fail too, so it belongs
inside the try.
Hint 2: a first step
Start without nesting: begin, yield connection, _finish(...), and on
BaseException, _rollback(...) then raise. Get the plain commit and
rollback tests passing, then add the depth bookkeeping for savepoints, then
_translate, then the guard around the rollback that the TODO describes.
Show one solution
@contextmanagerdef transaction(connection: Connection) -> Iterator[Connection]: 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: error.add_note( # keep it, don't raise it 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)
The structure carries the whole lesson. _finish — the commit, or a savepoint
release — is inside the try, because a commit is exactly when a database
checks deferred constraints, so it can fail. A commit that raises is caught
here, rolled back, and translated like any other failure. Put _finish after
the try (in an else) and a failed commit escapes untranslated, leaving the
connection stuck in a transaction nobody can end. There is a whole test for that.
Inside the handler the order is roll back first, translate second. If the
translation raised before the rollback, the transaction would stay open and the
next statement on that connection would join a transaction nobody owns.
The rollback is guarded, and its failure becomes a note. The rollback can
fail for a dull reason: SQLite’s INSERT OR ROLLBACK ends the transaction
itself, so by the time we try, there is nothing to roll back and we get
OperationalError: cannot rollback - no transaction is active. Let that
propagate and the caller sees the cleanup’s complaint instead of the unique-
constraint violation that actually happened — the exact failure mode Day 4.3
warns about. But swallowing it with a bare pass goes too far the other way: a
rollback that fails for a real reason means the connection may be unusable,
and that is worth knowing. add_note is the middle path. It attaches the
secondary error to the primary one, so the traceback shows both and the caller
still catches the error they expected.
sqlite3.IntegrityError: UNIQUE constraint failed: post.titlerollback after this error also failed: OperationalError('cannot rollback - no transaction is active')
One consequence is easy to miss: because _translate raises a new exception,
it has to copy the notes across, or the one you just attached is lost at the
moment of translation. Two mutations in the manifest — one removing the
add_note, one dropping the copy — both turn the suite red.
_translate uses isinstance(error, errors.DRIVER_INTEGRITY_ERRORS) rather
than an except errors.DRIVER_INTEGRITY_ERRORS clause. Not because of when
the tuple is read — Python evaluates an except expression each time the clause
is reached, so an except clause would see the registered value too — but
because the tuple starts empty (except () would be dead until a driver
registers), and one broad handler that checks the current tuple reads more
plainly than a clause whose meaning depends on import order.
finally restores the previous depth rather than decrementing: after an
exception, “what it was before” is the only value that is definitely right.
Bug hunt: a promise that doesn’t scale
save() is all-or-nothing. So someone wrote save_all() the obvious way:
# Each save() is already all-or-nothing, so saving them one by one # gives the same guarantee for the batch. for obj in objects: obj.save(connection)
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.
"""
from collections.abc import Iterable
from typing import Any, ClassVar, Self, cast
from warnings import deprecated
from errors import NotFound
from fields import Field, IntegerField, TextField
from protocols import Connection
from transactions import transaction
class Model:
#: Every model, by table name.
registry: ClassVar[dict[str, type["Model"]]] = {}
table: ClassVar[str]
fields: ClassVar[dict[str, Field[Any]]]
_instances: ClassVar[dict[int, Any]]
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)
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
with transaction(connection):
self._write(connection)
self._remember() # only once the write is committed
return self
@classmethod
def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None:
"""Write several objects in one transaction: all of them, or none."""
# Each save() is already all-or-nothing, so saving them one by one
# gives the same guarantee for the batch.
for obj in objects:
obj.save(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]
)
def _remember(self) -> None:
"""Put this object in its model's identity map."""
assert self.id is not None
type(self)._instances[self.id] = self
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cached = cls._instances.get(id)
if cached is not None:
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()})
cls._instances[id] = found
return found
@classmethod
@deprecated("Model.fetch is deprecated; use Model.get instead")
def fetch(cls, connection: Connection, id: int) -> Self:
return cls.get(connection, id)
def __repr__(self) -> str:
shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields)
return f"{type(self).__name__}(id={self.id!r}, {shown})"
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
class User(Model, table="user"):
email = TextField(max_length=200)
"""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
"""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,)
"""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
"""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: ...
@runtime_checkable
class Connection(Protocol):
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: ...
def commit(self) -> None: ...
"""SQLite scaffolding, supplied by the course.
`isolation_level=None` turns off the driver's own implicit transactions, so the
library's `transaction()` is the only thing issuing begin, commit and rollback.
"""
import sqlite3
import errors
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null unique,
views integer not null default 0
);
create table if not exists user (
id integer primary key autoincrement,
email text not null
);
"""
def connect(path: str = ":memory:") -> sqlite3.Connection:
connection = sqlite3.connect(path, isolation_level=None)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
errors.use_driver(integrity=sqlite3.IntegrityError)
return connection
import pathlib
import sqlite3
import tempfile
import unittest
import errors
import transactions
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
from transactions import transaction
class Hierarchy(unittest.TestCase):
def test_every_error_is_an_orm_error(self) -> None:
for error in (ValidationError, NotFound, IntegrityError):
self.assertTrue(issubclass(error, ORMError), error.__name__)
def test_errors_keep_their_builtin_meaning(self) -> None:
self.assertTrue(issubclass(ValidationError, ValueError))
self.assertTrue(issubclass(NotFound, LookupError))
def test_a_field_raises_validation_error(self) -> None:
with self.assertRaises(ValidationError):
Post(title="Hello", views=-1)
def test_old_code_catching_valueerror_still_works(self) -> None:
try:
Post(title="Hello", views=-1)
except ValueError as error:
self.assertIsInstance(error, ValidationError)
else:
self.fail("no error raised")
class Transactions(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def rows(self) -> list[str]:
return [row["title"] for row in self.connection.execute("select title from post")]
def test_save_all_is_all_or_nothing(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="second", views=1),
Post(title="first", views=2)])
self.assertEqual(self.rows(), ["first"], "the good row was rolled back too")
def test_the_driver_error_is_the_cause(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError) as caught:
Post(title="first", views=2).save(self.connection)
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
def test_any_exception_rolls_back(self) -> None:
with self.assertRaises(ZeroDivisionError):
with transaction(self.connection):
Post(title="doomed", views=1)._write(self.connection)
1 / 0
self.assertEqual(self.rows(), [])
def test_a_nested_failure_keeps_the_outer_work(self) -> None:
with transaction(self.connection):
Post(title="outer", views=1)._write(self.connection)
try:
with transaction(self.connection):
Post(title="inner", views=1)._write(self.connection)
raise RuntimeError("no good")
except RuntimeError:
pass
self.assertEqual(self.rows(), ["outer"])
def test_a_failed_save_is_not_cached(self) -> None:
Post(title="first", views=1).save(self.connection)
doomed = Post(title="first", views=2)
with self.assertRaises(IntegrityError):
doomed.save(self.connection)
self.assertNotIn(doomed, Post._instances.values())
def test_nothing_is_left_behind_between_transactions(self) -> None:
Post(title="first", views=1).save(self.connection)
self.assertEqual(transactions._depth, {}, "a connection stayed in the depth map")
def test_a_failure_during_commit_rolls_back_and_recovers(self) -> None:
"""A deferred constraint fails at COMMIT, not at the offending statement."""
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table parent (id integer primary key)")
self.connection.execute(
"create table child (parent_id integer references parent(id) "
"deferrable initially deferred)"
)
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute("insert into child values (999)")
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
# The connection must be usable afterwards, not stuck mid-transaction.
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
with transaction(self.connection):
self.connection.execute("insert into parent values (1)")
self.assertEqual(self.connection.execute("select count(*) from parent").fetchone()[0], 1)
def test_a_conflict_clause_that_self_rolls_back_still_reports_our_error(self) -> None:
"""INSERT OR ROLLBACK ends the transaction itself, so our rollback can't
run — but the caller must still get IntegrityError, not the secondary
'no transaction is active' OperationalError."""
self.connection.execute("insert into post (title, views) values ('same', 0)")
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute(
"insert or rollback into post (title, views) values ('same', 1)")
# the cleanup's own failure is recorded as a note, not raised in place
# of the real error -- and the translation carries the note across
self.assertTrue(any("rollback after this error also failed" in note
for note in getattr(caught.exception, "__notes__", ())))
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
# and the connection is still usable
with transaction(self.connection):
self.connection.execute("insert into post (title, views) values ('other', 2)")
self.assertEqual(sorted(self.rows()), ["other", "same"])
def test_a_commit_really_reaches_the_file(self) -> None:
"""An in-memory database can't tell a commit from an open transaction."""
with tempfile.TemporaryDirectory() as folder:
path = str(pathlib.Path(folder) / "posts.db")
first = connect(path)
Post(title="kept", views=1).save(first)
first.close()
second = connect(path)
titles = [row["title"] for row in second.execute("select title from post")]
second.close()
self.assertEqual(titles, ["kept"])
class StillWorks(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_round_trip(self) -> None:
post = Post(title="Hello", views=3).save(self.connection)
self.assertEqual(Post.get(self.connection, post.id).views, 3)
def test_missing_rows_raise_not_found(self) -> None:
with self.assertRaises(NotFound):
Post.get(self.connection, 99)
def test_two_models_stay_apart(self) -> None:
Post(title="Hello", views=0).save(self.connection)
User(email="a@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Reproduce. One failure:
AssertionError: Lists differ: ['first', 'second'] != ['first']: the good row was rolled back too
The test saved a valid post and a duplicate. The duplicate raised, as it
should. 'second' is still in the table.
Inspect. The comment in the code is the bug, written out in English. Each
save() opens its own transaction and commits it. By the time the second
object fails, the first one’s transaction is finished — there is nothing left
to roll back.
Narrow. “All or nothing” is a property of a unit of work, and the unit
here is the batch, not the row. Composing two atomic operations does not give
you an atomic operation; it gives you two, and a gap in between.
Pin.test_save_all_is_all_or_nothing is the regression test, and its
message says what the caller lost: the good row was rolled back too is the
behaviour being asserted.
Fix. One transaction around the whole batch:
written = list(objects) with transaction(connection): for obj in written: obj._write(connection) for obj in written: obj._remember()
Nesting is what makes this possible: _write does the SQL, and save() is
just _write wrapped in a transaction of its own. If save_all called
save() inside its own with, the savepoint machinery would handle it
correctly too — but doing the work once, at the right level, is clearer.
Why list(objects) first
objects is an Iterable, so it might be a generator. Consuming it inside the
with block and then iterating again afterwards to update the cache would find
it empty. Materialising it once, before the transaction opens, also means a
generator that raises does so outside the transaction.
The cache and the commit
Notice where _remember() happens: after the with block, never inside it.
def save(self, connection: Connection) -> Self: with transaction(connection): self._write(connection) self._remember() # only once the write is committed return self
Put it inside, and a rolled-back object stays in the identity map — so
Post.get(connection, 1) returns an object that doesn’t exist in the database,
from memory, without ever running a query. That’s the worst kind of bug: no
exception, no bad data on disk, just a library confidently answering with
something imaginary.
There is a deeper version of this problem that the library does not solve:
an object saved successfully inside an outer transaction that later rolls back
is cached, wrongly. Fixing that means the cache has to participate in the
transaction — deferring its updates until the outermost commit. It’s a good
exercise, and it’s the sort of thing Day 7’s write-up should name as a known
limitation rather than leave for a user to discover.
Where Day 4 leaves the library
The finished Day 4Python"""The base model, typed.
`get` returns `Self`, so `Post.get(...)` is a `Post` to a checker, not a
`Model`. The connection is a `Protocol`, so the library never imports sqlite3.
"""
from collections.abc import Iterable
from typing import Any, ClassVar, Self, cast
from warnings import deprecated
from errors import NotFound
from fields import Field, IntegerField, TextField
from protocols import Connection
from transactions import transaction
class Model:
#: Every model, by table name.
registry: ClassVar[dict[str, type["Model"]]] = {}
table: ClassVar[str]
fields: ClassVar[dict[str, Field[Any]]]
_instances: ClassVar[dict[int, Any]]
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)
def save(self, connection: Connection) -> Self:
"""Write this object. All of it, or none of it."""
with transaction(connection):
self._write(connection)
self._remember() # only once the write is committed
return self
@classmethod
def save_all(cls, connection: Connection, objects: Iterable["Model"]) -> None:
"""Write several objects in one transaction: all of them, or none."""
written = list(objects)
with transaction(connection):
for obj in written:
obj._write(connection)
for obj in written:
obj._remember()
def _write(self, connection: Connection) -> None:
columns = self.columns
values = [getattr(self, name) for name in columns]
names = ", ".join(columns)
if self.id is None:
marks = ", ".join("?" for _ in columns)
cursor = connection.execute(
f"insert into {self.table} ({names}) values ({marks})", values
)
self.id = cursor.lastrowid
else:
assignments = ", ".join(f"{name} = ?" for name in columns)
connection.execute(
f"update {self.table} set {assignments} where id = ?", [*values, self.id]
)
def _remember(self) -> None:
"""Put this object in its model's identity map."""
assert self.id is not None
type(self)._instances[self.id] = self
@classmethod
def get(cls, connection: Connection, id: int) -> Self:
cached = cls._instances.get(id)
if cached is not None:
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()})
cls._instances[id] = found
return found
@classmethod
@deprecated("Model.fetch is deprecated; use Model.get instead")
def fetch(cls, connection: Connection, id: int) -> Self:
return cls.get(connection, id)
def __repr__(self) -> str:
shown = ", ".join(f"{name}={getattr(self, name, None)!r}" for name in self.fields)
return f"{type(self).__name__}(id={self.id!r}, {shown})"
class Post(Model, table="post"):
title = TextField(max_length=100)
views = IntegerField(minimum=0)
class User(Model, table="user"):
email = TextField(max_length=200)
"""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
"""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,)
"""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
"""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: ...
@runtime_checkable
class Connection(Protocol):
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: ...
def commit(self) -> None: ...
"""SQLite scaffolding, supplied by the course.
`isolation_level=None` turns off the driver's own implicit transactions, so the
library's `transaction()` is the only thing issuing begin, commit and rollback.
"""
import sqlite3
import errors
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null unique,
views integer not null default 0
);
create table if not exists user (
id integer primary key autoincrement,
email text not null
);
"""
def connect(path: str = ":memory:") -> sqlite3.Connection:
connection = sqlite3.connect(path, isolation_level=None)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
errors.use_driver(integrity=sqlite3.IntegrityError)
return connection
import pathlib
import sqlite3
import tempfile
import unittest
import errors
import transactions
from db import connect
from errors import IntegrityError, NotFound, ORMError, ValidationError
from models import Model, Post, User
from transactions import transaction
class Hierarchy(unittest.TestCase):
def test_every_error_is_an_orm_error(self) -> None:
for error in (ValidationError, NotFound, IntegrityError):
self.assertTrue(issubclass(error, ORMError), error.__name__)
def test_errors_keep_their_builtin_meaning(self) -> None:
self.assertTrue(issubclass(ValidationError, ValueError))
self.assertTrue(issubclass(NotFound, LookupError))
def test_a_field_raises_validation_error(self) -> None:
with self.assertRaises(ValidationError):
Post(title="Hello", views=-1)
def test_old_code_catching_valueerror_still_works(self) -> None:
try:
Post(title="Hello", views=-1)
except ValueError as error:
self.assertIsInstance(error, ValidationError)
else:
self.fail("no error raised")
class Transactions(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def rows(self) -> list[str]:
return [row["title"] for row in self.connection.execute("select title from post")]
def test_save_all_is_all_or_nothing(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError):
Post.save_all(self.connection, [Post(title="second", views=1),
Post(title="first", views=2)])
self.assertEqual(self.rows(), ["first"], "the good row was rolled back too")
def test_the_driver_error_is_the_cause(self) -> None:
Post(title="first", views=1).save(self.connection)
with self.assertRaises(IntegrityError) as caught:
Post(title="first", views=2).save(self.connection)
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
def test_any_exception_rolls_back(self) -> None:
with self.assertRaises(ZeroDivisionError):
with transaction(self.connection):
Post(title="doomed", views=1)._write(self.connection)
1 / 0
self.assertEqual(self.rows(), [])
def test_a_nested_failure_keeps_the_outer_work(self) -> None:
with transaction(self.connection):
Post(title="outer", views=1)._write(self.connection)
try:
with transaction(self.connection):
Post(title="inner", views=1)._write(self.connection)
raise RuntimeError("no good")
except RuntimeError:
pass
self.assertEqual(self.rows(), ["outer"])
def test_a_failed_save_is_not_cached(self) -> None:
Post(title="first", views=1).save(self.connection)
doomed = Post(title="first", views=2)
with self.assertRaises(IntegrityError):
doomed.save(self.connection)
self.assertNotIn(doomed, Post._instances.values())
def test_nothing_is_left_behind_between_transactions(self) -> None:
Post(title="first", views=1).save(self.connection)
self.assertEqual(transactions._depth, {}, "a connection stayed in the depth map")
def test_a_failure_during_commit_rolls_back_and_recovers(self) -> None:
"""A deferred constraint fails at COMMIT, not at the offending statement."""
self.connection.execute("pragma foreign_keys=on")
self.connection.execute("create table parent (id integer primary key)")
self.connection.execute(
"create table child (parent_id integer references parent(id) "
"deferrable initially deferred)"
)
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute("insert into child values (999)")
self.assertIsInstance(caught.exception.__cause__, sqlite3.IntegrityError)
# The connection must be usable afterwards, not stuck mid-transaction.
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
with transaction(self.connection):
self.connection.execute("insert into parent values (1)")
self.assertEqual(self.connection.execute("select count(*) from parent").fetchone()[0], 1)
def test_a_conflict_clause_that_self_rolls_back_still_reports_our_error(self) -> None:
"""INSERT OR ROLLBACK ends the transaction itself, so our rollback can't
run — but the caller must still get IntegrityError, not the secondary
'no transaction is active' OperationalError."""
self.connection.execute("insert into post (title, views) values ('same', 0)")
with self.assertRaises(IntegrityError) as caught:
with transaction(self.connection):
self.connection.execute(
"insert or rollback into post (title, views) values ('same', 1)")
# the cleanup's own failure is recorded as a note, not raised in place
# of the real error -- and the translation carries the note across
self.assertTrue(any("rollback after this error also failed" in note
for note in getattr(caught.exception, "__notes__", ())))
self.assertFalse(self.connection.in_transaction)
self.assertEqual(transactions._depth, {})
# and the connection is still usable
with transaction(self.connection):
self.connection.execute("insert into post (title, views) values ('other', 2)")
self.assertEqual(sorted(self.rows()), ["other", "same"])
def test_a_commit_really_reaches_the_file(self) -> None:
"""An in-memory database can't tell a commit from an open transaction."""
with tempfile.TemporaryDirectory() as folder:
path = str(pathlib.Path(folder) / "posts.db")
first = connect(path)
Post(title="kept", views=1).save(first)
first.close()
second = connect(path)
titles = [row["title"] for row in second.execute("select title from post")]
second.close()
self.assertEqual(titles, ["kept"])
class StillWorks(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_round_trip(self) -> None:
post = Post(title="Hello", views=3).save(self.connection)
self.assertEqual(Post.get(self.connection, post.id).views, 3)
def test_missing_rows_raise_not_found(self) -> None:
with self.assertRaises(NotFound):
Post.get(self.connection, 99)
def test_two_models_stay_apart(self) -> None:
Post(title="Hello", views=0).save(self.connection)
User(email="a@example.com").save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
self.assertIsInstance(User.get(self.connection, 1), User)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
One test in there is worth reading before you go, because it tests the test
setup as much as the library:
def test_a_commit_really_reaches_the_file(self) -> None: """An in-memory database can't tell a commit from an open transaction."""
Every other test uses :memory:, where an uncommitted transaction looks exactly
like a committed one from the same connection. One test with a real file, a
close and a re-open is what proves the commit happened at all.
What you learned
Atomic operations don’t compose: the unit of work is the batch, and it needs
a transaction of its own.
Roll back before translating an exception, or you leave the transaction open.
Cache after the commit, never before — an identity map that remembers a
rolled-back object answers from memory with something that never existed.
In-memory databases hide commit bugs; one file-based test covers that.
Tomorrow is a day away from the ORM: asyncio, where “clean up on the way out”
gets much harder because the way out might be a cancellation.
Your progress is saved in this browser.
Sign in
to keep it on every device.