The library is typed. Time to find out what that bought, and — more useful —
what it didn’t.
The checkpoint
protocols.py is empty except for the docstring. Write the three protocols the
library needs, and twelve runtime tests will tell you whether they describe the
real sqlite3 connection, a hand-written substitute, and an object that is
missing a method.
Checkpoint
Day 3 checkpoint: describe the connection
protocols.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 typing import Any, ClassVar, Self, cast
from warnings import deprecated
from fields import Field, IntegerField, TextField
from protocols import Connection
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:
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]
)
connection.commit()
assert self.id is not None
type(self)._instances[self.id] = self
return 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 LookupError(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)
"""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
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 ValueError.
`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 ValueError(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 ValueError(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 ValueError(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 ValueError(f"{self.name} must be an int, not {type(value).__name__}")
if self.minimum is not None and value < self.minimum:
raise ValueError(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 the library *uses*, so anything with those methods fits: a
real connection, a wrapper that logs, a fake in a test.
TODO: write the three protocols the tests describe.
Row keys() -> Sequence[str], and row[column] -> Any
Cursor a read-only lastrowid (int | None), and fetchone() -> Row | None
Connection execute(sql, parameters=(), /) -> Cursor, and commit() -> None
Make Cursor and Connection @runtime_checkable, so isinstance works.
Mind how you declare lastrowid: a plain annotation in a Protocol means the
attribute must be readable *and* writable, and sqlite3's is read-only.
"""
from collections.abc import Sequence
from typing import Any, Protocol, runtime_checkable
class Row(Protocol):
...
class Cursor(Protocol):
...
class Connection(Protocol):
...
"""SQLite scaffolding, supplied by the course.
Day 1 is about Python's object model, not about SQL, so this file is given to
you: a connection with the library's two tables already created. The rest of
the week replaces almost everything else, but never this.
"""
import sqlite3
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null,
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:
"""A connection whose rows behave like dictionaries."""
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
return connection
import sqlite3
import unittest
from collections.abc import Sequence
from typing import Any
import warnings
from db import connect
from fields import TextField
from models import Model, Post, User
from protocols import Connection
class RecordingConnection:
"""Not a subclass of anything: it just has the methods the library uses."""
def __init__(self) -> None:
self.statements: list[str] = []
self.real = sqlite3.connect(":memory:")
self.real.row_factory = sqlite3.Row
self.real.executescript(
"create table post (id integer primary key autoincrement,"
" title text not null, views integer not null default 0)"
)
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
self.statements.append(sql)
return self.real.execute(sql, parameters)
def commit(self) -> None:
self.real.commit()
class StructuralTyping(unittest.TestCase):
def test_a_real_connection_satisfies_the_protocol(self) -> None:
self.assertIsInstance(connect(), Connection)
def test_so_does_anything_with_the_right_methods(self) -> None:
self.assertIsInstance(RecordingConnection(), Connection)
def test_something_missing_a_method_does_not(self) -> None:
class NoCommit:
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
raise NotImplementedError
self.assertNotIsInstance(NoCommit(), Connection)
def test_isinstance_ignores_signatures(self) -> None:
"""runtime_checkable checks that the names exist, nothing more."""
class WrongSignatures:
def execute(self) -> None: ...
def commit(self, unexpected: int) -> None: ...
self.assertIsInstance(WrongSignatures(), Connection)
def test_the_library_works_with_a_substitute(self) -> None:
connection = RecordingConnection()
post = Post(title="Hello", views=1).save(connection)
assert post.id is not None
Post._instances.clear()
loaded = Post.get(connection, post.id)
self.assertEqual(loaded.title, "Hello")
self.assertTrue(any(sql.startswith("insert into post") for sql in connection.statements))
class TypedApi(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_get_returns_the_subclass(self) -> None:
Post(title="Hello", views=1).save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
def test_fetch_is_deprecated_but_still_works(self) -> None:
Post(title="Hello", views=1).save(self.connection)
with self.assertWarns(DeprecationWarning):
post = Post.fetch(self.connection, 1) # type: ignore[deprecated] # that is what we are testing
self.assertEqual(post.title, "Hello")
def test_values_from_the_database_are_validated_too(self) -> None:
"""A row written behind the library's back still goes through the fields."""
self.connection.execute("insert into post (title, views) values ('x', -5)")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_a_text_value_in_an_integer_column_never_reaches_the_object(self) -> None:
"""SQLite stores what it is given; the field is what keeps int an int."""
self.connection.execute("insert into post (title, views) values ('x', 'lots')")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_unknown_columns_are_still_refused(self) -> None:
with self.assertRaises(TypeError):
Post(title="Hello", author="nobody")
class Fields(unittest.TestCase):
def test_validation_still_applies(self) -> None:
post = Post(title="Hello", views=0)
with self.assertRaises(ValueError):
post.views = "lots" # type: ignore[assignment] # the point of the test
self.assertEqual(post.views, 0)
def test_a_null_field_accepts_none(self) -> None:
class Draft(Model, table="draft"):
note = TextField(null=True)
draft = Draft()
draft.note = None # type: ignore[assignment] # null=True, invisible to the checker
self.assertIsNone(draft.note)
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
Describe exactly what the library calls on a connection, a cursor and a row —
no more. The twelve tests check names only, so loose declarations pass them.
The precise parts, like a read-only lastrowid, are what make the real
sqlite3 classes match under mypy, which the next section is about.
Hint 2: a first step
Read models.py and list every method and attribute it uses on each of the
three objects: that list is the three protocols. Connection is the shortest —
execute and commit — so start there.
Show one solution
@runtime_checkableclass Row(Protocol): def keys(self) -> Sequence[str]: ... def __getitem__(self, key: str, /) -> Any: ...@runtime_checkableclass 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_checkableclass Connection(Protocol): def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> Cursor: ... def commit(self) -> None: ...
The tests here check only what isinstance can: the names. The @property on
lastrowid and the / in execute make no difference at run time at all —
every test passes without them. They matter to mypy, and the next section is
where you find that out.
The exact diagnostics depend on the checker, so the quoted output is from
mypy 2.3 targeting Python 3.14 (these files need 3.13+ for
warnings.deprecated). Pin the version and the target when you quote a
checker’s output, or a reader on a different mypy sees different messages.
Five errors, all in misuse.py:
misuse.py:12: error: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment]misuse.py:13: error: Incompatible types in assignment (expression has type "int", variable has type "str") [assignment]misuse.py:14: error: Incompatible types in assignment (expression has type "str", variable has type "int") [assignment]misuse.py:15: error: function models.Model.fetch is deprecated: Model.fetch is deprecated; use Model.get instead [deprecated]misuse.py:16: error: "Post" has no attribute "author" [attr-defined]
misuse.py is a file of deliberate mistakes, and each line is a class of bug
the checker now catches for your users before they run anything: assigning the
wrong type to a field, reading a field as the wrong type, calling something
deprecated, and inventing a column that doesn’t exist.
Then try breaking the library on purpose. Change lastrowid from a property
back to a plain attribute and re-run: every call site that passes a real
sqlite3.Connection becomes an error, while all twelve tests keep passing. A
protocol that is too demanding is invisible to tests — it’s the clearest case in
this course for a checker earning its place in CI.
Why the tests are annotated too
Test files are where over-strict types show up first, and where type: ignore
gets used honestly: post.views = "lots" # type: ignore[assignment] in a test
that asserts ValueError is exactly right. The comment marks the line as
deliberately wrong, which is what it is.
Bug hunt: the fast path that let a string in
Someone optimised get(). Rows come from our own tables, they reasoned, so
skipping __init__ and the descriptors saves a few calls per row.
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 typing import Any, ClassVar, Self, cast
from warnings import deprecated
from fields import Field, IntegerField, TextField
from protocols import Connection
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:
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]
)
connection.commit()
assert self.id is not None
type(self)._instances[self.id] = self
return 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 LookupError(f"no {cls.__name__} with id {id}")
# Rows come from our own tables, so skip __init__ and the fields and
# drop the values straight in. Fewer function calls per row.
found = cls.__new__(cls)
found.id = row["id"]
for key in row.keys():
if key != "id":
found.__dict__[f"_{key}"] = row[key]
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)
"""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
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 ValueError.
`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 ValueError(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 ValueError(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 ValueError(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 ValueError(f"{self.name} must be an int, not {type(value).__name__}")
if self.minimum is not None and value < self.minimum:
raise ValueError(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.
Day 1 is about Python's object model, not about SQL, so this file is given to
you: a connection with the library's two tables already created. The rest of
the week replaces almost everything else, but never this.
"""
import sqlite3
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null,
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:
"""A connection whose rows behave like dictionaries."""
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
return connection
import sqlite3
import unittest
from collections.abc import Sequence
from typing import Any
import warnings
from db import connect
from fields import TextField
from models import Model, Post, User
from protocols import Connection
class RecordingConnection:
"""Not a subclass of anything: it just has the methods the library uses."""
def __init__(self) -> None:
self.statements: list[str] = []
self.real = sqlite3.connect(":memory:")
self.real.row_factory = sqlite3.Row
self.real.executescript(
"create table post (id integer primary key autoincrement,"
" title text not null, views integer not null default 0)"
)
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
self.statements.append(sql)
return self.real.execute(sql, parameters)
def commit(self) -> None:
self.real.commit()
class StructuralTyping(unittest.TestCase):
def test_a_real_connection_satisfies_the_protocol(self) -> None:
self.assertIsInstance(connect(), Connection)
def test_so_does_anything_with_the_right_methods(self) -> None:
self.assertIsInstance(RecordingConnection(), Connection)
def test_something_missing_a_method_does_not(self) -> None:
class NoCommit:
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
raise NotImplementedError
self.assertNotIsInstance(NoCommit(), Connection)
def test_isinstance_ignores_signatures(self) -> None:
"""runtime_checkable checks that the names exist, nothing more."""
class WrongSignatures:
def execute(self) -> None: ...
def commit(self, unexpected: int) -> None: ...
self.assertIsInstance(WrongSignatures(), Connection)
def test_the_library_works_with_a_substitute(self) -> None:
connection = RecordingConnection()
post = Post(title="Hello", views=1).save(connection)
assert post.id is not None
Post._instances.clear()
loaded = Post.get(connection, post.id)
self.assertEqual(loaded.title, "Hello")
self.assertTrue(any(sql.startswith("insert into post") for sql in connection.statements))
class TypedApi(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_get_returns_the_subclass(self) -> None:
Post(title="Hello", views=1).save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
def test_fetch_is_deprecated_but_still_works(self) -> None:
Post(title="Hello", views=1).save(self.connection)
with self.assertWarns(DeprecationWarning):
post = Post.fetch(self.connection, 1) # type: ignore[deprecated] # that is what we are testing
self.assertEqual(post.title, "Hello")
def test_values_from_the_database_are_validated_too(self) -> None:
"""A row written behind the library's back still goes through the fields."""
self.connection.execute("insert into post (title, views) values ('x', -5)")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_a_text_value_in_an_integer_column_never_reaches_the_object(self) -> None:
"""SQLite stores what it is given; the field is what keeps int an int."""
self.connection.execute("insert into post (title, views) values ('x', 'lots')")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_unknown_columns_are_still_refused(self) -> None:
with self.assertRaises(TypeError):
Post(title="Hello", author="nobody")
class Fields(unittest.TestCase):
def test_validation_still_applies(self) -> None:
post = Post(title="Hello", views=0)
with self.assertRaises(ValueError):
post.views = "lots" # type: ignore[assignment] # the point of the test
self.assertEqual(post.views, 0)
def test_a_null_field_accepts_none(self) -> None:
class Draft(Model, table="draft"):
note = TextField(null=True)
draft = Draft()
draft.note = None # type: ignore[assignment] # null=True, invisible to the checker
self.assertIsNone(draft.note)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
Reproduce. Two failures, both saying ValueError not raised:
self.connection.execute("insert into post (title, views) values ('x', 'lots')") self.connection.commit() with self.assertRaises(ValueError): Post.get(self.connection, 1)
Inspect. Add a line to see what you got instead of an exception:
post = Post.get(self.connection, 1) print(repr(post.views), type(post.views))
'lots' <class 'str'> — in a field declared IntegerField(minimum=0), on an
attribute every annotation in the library says is an int. The next caller to
write post.views + 1 gets a TypeError far from here.
Narrow. The new get() writes found.__dict__["_views"] directly. That’s
the storage slot behind the descriptor, so __set__ never runs and check()
never runs. The fast path didn’t skip “a few calls” — it skipped the validation.
Why SQLite let that in at all
The column is declared integer, but SQLite has type affinity, not type
enforcement: a value it can’t convert is stored as-is. So 'lots' sits happily
in an integer column. Your library is the only thing standing between that and
your caller’s arithmetic — which is the answer to “why validate on load, we
wrote the row ourselves?”.
Pin. The two failing tests are the regression tests, and one of them says
the important thing in its name: a text value in an integer column never
reaches the object.
Fix. Go through the fields, the same way a caller’s assignment does:
found = cls(**{key: row[key] for key in row.keys()})
What the checker can’t see
Four honest limits, all of them live in this library:
Data from outside. Everything crossing a boundary — a database, JSON, a
request — is Any at the door. The annotation on post.views describes what
the code believes; only the runtime check makes it true. Types and validation
aren’t alternatives, they’re two halves, and the bug hunt above is what happens
when you have only one.
null=True.TextField(null=True) accepts None at run time, and
Field[str] still reports str. The checker will happily let you call
.upper() on it:
…which is reported even though it’s allowed. This is not an argument
that nullable fields can’t be typed — they can. A separate Field[str | None]
type (or a distinct NullableTextField) would express it soundly, at the cost
of two field classes instead of one. What doesn’t work is trying to make a
single TextField flip its element type on a null=True flag via @overloaded
__new__: that technique quietly breaks as soon as the class also defines
__init__, which mypy prefers for inference (verified). This library keeps the
single-class API and swallows the unsoundness, and the cast(str, None) in each
check is therefore a knowingly false annotation — not the justified kind
of cast (reading back a value a subclass already validated), but a deliberate
hole the null flag opens. If you’d rather stay sound, split the field types.
cast and type: ignore. Every one is a place where you told the checker
to stop looking. The library has five casts — four in fields.py, one in
models.py — and its tests add three narrow ignores. Each has a reason, and
together they’re the honest footnote on “it type-checks”. Count them in a
codebase you inherit: it’s a good measure of what the green tick is worth.
Everything about behaviour. Types can’t say that save() commits, that
get() returns a cached object rather than a fresh read, or that assigning
views rejects negatives. A checker proves shapes fit together. Tests prove the
code does what it claims. You need both, and neither substitutes for the other.
Where Day 3 leaves the library
The finished Day 3Python"""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 typing import Any, ClassVar, Self, cast
from warnings import deprecated
from fields import Field, IntegerField, TextField
from protocols import Connection
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:
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]
)
connection.commit()
assert self.id is not None
type(self)._instances[self.id] = self
return 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 LookupError(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)
"""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
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 ValueError.
`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 ValueError(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 ValueError(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 ValueError(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 ValueError(f"{self.name} must be an int, not {type(value).__name__}")
if self.minimum is not None and value < self.minimum:
raise ValueError(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.
Day 1 is about Python's object model, not about SQL, so this file is given to
you: a connection with the library's two tables already created. The rest of
the week replaces almost everything else, but never this.
"""
import sqlite3
SCHEMA = """
create table if not exists post (
id integer primary key autoincrement,
title text not null,
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:
"""A connection whose rows behave like dictionaries."""
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.executescript(SCHEMA)
return connection
import sqlite3
import unittest
from collections.abc import Sequence
from typing import Any
import warnings
from db import connect
from fields import TextField
from models import Model, Post, User
from protocols import Connection
class RecordingConnection:
"""Not a subclass of anything: it just has the methods the library uses."""
def __init__(self) -> None:
self.statements: list[str] = []
self.real = sqlite3.connect(":memory:")
self.real.row_factory = sqlite3.Row
self.real.executescript(
"create table post (id integer primary key autoincrement,"
" title text not null, views integer not null default 0)"
)
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
self.statements.append(sql)
return self.real.execute(sql, parameters)
def commit(self) -> None:
self.real.commit()
class StructuralTyping(unittest.TestCase):
def test_a_real_connection_satisfies_the_protocol(self) -> None:
self.assertIsInstance(connect(), Connection)
def test_so_does_anything_with_the_right_methods(self) -> None:
self.assertIsInstance(RecordingConnection(), Connection)
def test_something_missing_a_method_does_not(self) -> None:
class NoCommit:
def execute(self, sql: str, parameters: Sequence[Any] = (), /) -> sqlite3.Cursor:
raise NotImplementedError
self.assertNotIsInstance(NoCommit(), Connection)
def test_isinstance_ignores_signatures(self) -> None:
"""runtime_checkable checks that the names exist, nothing more."""
class WrongSignatures:
def execute(self) -> None: ...
def commit(self, unexpected: int) -> None: ...
self.assertIsInstance(WrongSignatures(), Connection)
def test_the_library_works_with_a_substitute(self) -> None:
connection = RecordingConnection()
post = Post(title="Hello", views=1).save(connection)
assert post.id is not None
Post._instances.clear()
loaded = Post.get(connection, post.id)
self.assertEqual(loaded.title, "Hello")
self.assertTrue(any(sql.startswith("insert into post") for sql in connection.statements))
class TypedApi(unittest.TestCase):
def setUp(self) -> None:
self.connection = connect()
for model in Model.registry.values():
model._instances.clear()
def test_get_returns_the_subclass(self) -> None:
Post(title="Hello", views=1).save(self.connection)
self.assertIsInstance(Post.get(self.connection, 1), Post)
def test_fetch_is_deprecated_but_still_works(self) -> None:
Post(title="Hello", views=1).save(self.connection)
with self.assertWarns(DeprecationWarning):
post = Post.fetch(self.connection, 1) # type: ignore[deprecated] # that is what we are testing
self.assertEqual(post.title, "Hello")
def test_values_from_the_database_are_validated_too(self) -> None:
"""A row written behind the library's back still goes through the fields."""
self.connection.execute("insert into post (title, views) values ('x', -5)")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_a_text_value_in_an_integer_column_never_reaches_the_object(self) -> None:
"""SQLite stores what it is given; the field is what keeps int an int."""
self.connection.execute("insert into post (title, views) values ('x', 'lots')")
self.connection.commit()
with self.assertRaises(ValueError):
Post.get(self.connection, 1)
def test_unknown_columns_are_still_refused(self) -> None:
with self.assertRaises(TypeError):
Post(title="Hello", author="nobody")
class Fields(unittest.TestCase):
def test_validation_still_applies(self) -> None:
post = Post(title="Hello", views=0)
with self.assertRaises(ValueError):
post.views = "lots" # type: ignore[assignment] # the point of the test
self.assertEqual(post.views, 0)
def test_a_null_field_accepts_none(self) -> None:
class Draft(Model, table="draft"):
note = TextField(null=True)
draft = Draft()
draft.note = None # type: ignore[assignment] # null=True, invisible to the checker
self.assertIsNone(draft.note)
import sys
import unittest
import test_models
suite = unittest.defaultTestLoader.loadTestsFromModule(test_models)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
"""Deliberate misuse. Every line below should be an error from `mypy .`,
and every one of them runs happily until it doesn't.
This file is not imported by the library or the tests: it exists to be checked.
"""
from db import connect
from models import Post
connection = connect()
post = Post(title="Hello", views=1)
post.views = "lots" # error: assignment of str to an int field
post.title = 42 # error: assignment of int to a str field
length: int = post.title # error: title is a str, not an int
Post.fetch(connection, 1) # error: deprecated
missing = post.author # error: Post has no attribute "author"
saved: Post = post.save(connection) # fine: save returns Self
loaded: Post = Post.get(connection, 1) # fine: get returns Self, not Model
Working on your own machine? Download the Day 3 files —
and this is the day to actually do it, because the checker is the point.
What you learned
Protocols written for real drivers need care about read-only members, and
only a checker tells you when you got it wrong.
mypy --strict over the library finds misuse in callers’ code before it runs.
A fast path that writes past a descriptor skips its validation, and makes
every annotation on that attribute a lie.
Typing and validation cover different ground: at every boundary, you need
the runtime check as well.
Tomorrow: what happens when things go wrong. The library gets its own exception
hierarchy, and save() learns to leave the database unchanged when it fails
halfway.
Your progress is saved in this browser.
Sign in
to keep it on every device.