build: check the core and source layers in pyright's strict mode

pyproject.toml

@@ -74,6 +74,8 @@ include = ["src"]
 pythonVersion = "3.11"
 pythonPlatform = "Linux"
 typeCheckingMode = "basic"
+# the configuration and source layers, and the schemas, are checked in strict mode
+strict = ["src/nnnotes/core", "src/nnnotes/source", "src/nnnotes/schemas"]
 
 [tool.importlinter]
 root_package = "nnnotes"

src/nnnotes/core/cache.py

@@ -28,12 +28,23 @@
 from collections import OrderedDict
 from importlib import metadata
 from pathlib import Path
-from typing import Any
+from typing import Any, Protocol, TypedDict, cast
 
 DISK_MAGIC = b"NNC1"
 DEFAULT_MEMORY_MB = 128
 
-_settings = {
+
+class Settings(TypedDict):
+    """The cache settings of a process (configure)."""
+
+    enabled: bool
+    memory: int  # byte budget of each bucket
+    directory: Path | None  # disk layer of the byte buckets
+    closures: int  # loaded bundle closures kept per thread
+    closure_bytes: int  # their bundle bytes
+
+
+_settings: Settings = {
     "enabled": True,
     "memory": DEFAULT_MEMORY_MB << 20,
     "directory": None,
@@ -48,10 +59,10 @@ def configure(
     *,
     enabled: bool | None = None,
     memory_mb: int | None = None,
-    directory=None,
+    directory: str | os.PathLike[str] | None = None,
     closures: int | None = None,
     closure_mb: int | None = None,
-) -> dict:
+) -> Settings:
     """Set the caches of this process: `enabled`; `memory_mb` (byte budget of each bucket); `directory` (disk layer
     of the byte buckets; "" or None for none); `closures` / `closure_mb` (loaded bundle closures kept by
     unity.load_closure, per thread: count and bundle bytes). Clears the buckets. Returns the settings."""
@@ -71,9 +82,9 @@ def configure(
         return settings()
 
 
-def settings() -> dict:
+def settings() -> Settings:
     """The current settings (a copy)."""
-    return dict(_settings)
+    return _settings.copy()
 
 
 def enabled() -> bool:
@@ -90,7 +101,7 @@ def bucket(name: str, *, salt: str = "", disk: bool = False, max_item: int | Non
         return b
 
 
-def stats() -> dict:
+def stats() -> dict[str, dict[str, int]]:
     """Per bucket: items, bytes, hits, misses, disk hits, disk writes, disk write errors."""
     return {n: b.stats() for n, b in sorted(_buckets.items())}
 
@@ -101,17 +112,22 @@ def clear() -> None:
 
 
 # ---------------------------------------------------------------- keys
-def _feed(h, part) -> None:
+class _Hash(Protocol):
+    def update(self, data: bytes | memoryview, /) -> None: ...
+
+
+def _feed(h: _Hash, part: object) -> None:
     if isinstance(part, (bytes, bytearray, memoryview)):
-        mv = memoryview(part).cast("B")
+        mv = memoryview(cast("bytes", part)).cast("B")  # any buffer, read as bytes
         h.update(b"b" + struct.pack("<Q", mv.nbytes))
         h.update(mv)
     elif isinstance(part, str):
         b = part.encode("utf-8", "surrogatepass")
         h.update(b"s" + struct.pack("<Q", len(b)) + b)
     elif isinstance(part, (tuple, list)):
-        h.update(b"t" + struct.pack("<Q", len(part)))
-        for p in part:
+        items = cast("list[object]", part)
+        h.update(b"t" + struct.pack("<Q", len(items)))
+        for p in items:
             _feed(h, p)
     elif part is None or isinstance(part, (bool, int, float)):
         r = repr(part).encode("ascii")
@@ -120,7 +136,7 @@ def _feed(h, part) -> None:
         raise TypeError(f"cache key part of type {type(part).__name__}")
 
 
-def key(*parts) -> str:
+def key(*parts: object) -> str:
     """sha256 hex of `parts` (bytes, str, int, float, bool, None and tuples / lists of them), unambiguous."""
     h = hashlib.sha256()
     _feed(h, parts)
@@ -130,14 +146,14 @@ def key(*parts) -> str:
 _salts: dict[str, str] = {}
 
 
-def source_salt(path, *dists: str) -> str:
+def source_salt(path: str | os.PathLike[str], *dists: str) -> str:
     """A salt for keys of values a module produces: sha256 of the module's source file and the installed versions
     of the distributions `dists` (a missing one counts as "-")."""
     p = Path(path)
     k = f"{p}|{'|'.join(dists)}"
     s = _salts.get(k)
     if s is None:
-        versions = []
+        versions: list[str] = []
         for d in dists:
             try:
                 versions.append(f"{d}={metadata.version(d)}")
@@ -147,7 +163,7 @@ def source_salt(path, *dists: str) -> str:
     return s
 
 
-def file_id(path) -> tuple:
+def file_id(path: str | os.PathLike[str]) -> tuple[str, int, int]:
     """(path, size, mtime) of a file: identifies an executable or input file in a key."""
     p = Path(path)
     st = p.stat()
@@ -156,14 +172,16 @@ def file_id(path) -> tuple:
 
 # ---------------------------------------------------------------- buckets
 class Bucket:
+    """A memo cache of values by content key (see the module documentation)."""
+
     def __init__(self, name: str, salt: str, disk: bool, max_item: int | None):
         self.name, self.salt, self.disk, self._max_item = name, salt, disk, max_item
-        self._items: OrderedDict[str, tuple] = OrderedDict()
+        self._items: OrderedDict[str, tuple[Any, int]] = OrderedDict()  # key -> (value, byte size)
         self._bytes = 0
         self._lock = threading.Lock()
         self.hits = self.misses = self.disk_hits = self.disk_writes = self.disk_errors = 0
 
-    def key(self, *parts) -> str:
+    def key(self, *parts: object) -> str:
         return key(self.name, self.salt, parts)
 
     def _limit(self) -> int:
@@ -180,8 +198,9 @@ def get(self, k: str) -> Any:
                 self._items.move_to_end(k)
                 self.hits += 1
                 return hit[0]
-        if self.disk and _settings["directory"] is not None:
-            data = _disk_read(self._path(k))
+        directory = _settings["directory"]
+        if self.disk and directory is not None:
+            data = _disk_read(self._path(directory, k))
             if data is not None:
                 with self._lock:
                     self.disk_hits += 1
@@ -191,21 +210,22 @@ def get(self, k: str) -> Any:
             self.misses += 1
         return None
 
-    def put(self, k: str, value, size: int | None = None) -> None:
+    def put(self, k: str, value: Any, size: int | None = None) -> None:
         """Store `value` (bytes for a disk bucket) under `k`; `size`: its byte size (default len(value))."""
         if not _settings["enabled"]:
             return
         n = len(value) if size is None else int(size)
         self._remember(k, value, n)
-        if self.disk and _settings["directory"] is not None and n <= max(self._limit(), 64 << 20):
-            ok = _disk_write(self._path(k), value)
+        directory = _settings["directory"]
+        if self.disk and directory is not None and n <= max(self._limit(), 64 << 20):
+            ok = _disk_write(self._path(directory, k), value)
             with self._lock:
                 if ok:
                     self.disk_writes += 1
                 else:
                     self.disk_errors += 1
 
-    def _remember(self, k: str, value, n: int) -> None:
+    def _remember(self, k: str, value: Any, n: int) -> None:
         if n > self._limit():
             return
         with self._lock:
@@ -219,15 +239,15 @@ def _remember(self, k: str, value, n: int) -> None:
                 _, (_, m) = self._items.popitem(last=False)
                 self._bytes -= m
 
-    def _path(self, k: str) -> Path:
-        return _settings["directory"] / self.name / k[:2] / k
+    def _path(self, directory: Path, k: str) -> Path:
+        return directory / self.name / k[:2] / k
 
     def clear(self) -> None:
         with self._lock:
             self._items.clear()
             self._bytes = 0
 
-    def stats(self) -> dict:
+    def stats(self) -> dict[str, int]:
         with self._lock:
             return {
                 "items": len(self._items),
@@ -298,7 +318,7 @@ def unpack(data: bytes) -> tuple[bytes, list[bytes]]:
     pos = 8
     meta = data[pos : pos + n]
     pos += n
-    blobs = []
+    blobs: list[bytes] = []
     for _ in range(count):
         (m,) = struct.unpack_from("<Q", data, pos)
         pos += 8

src/nnnotes/core/jsonio.py

@@ -14,20 +14,33 @@
 
 import json
 import math
+import os
+from collections.abc import Sequence
 from pathlib import Path
+from typing import Any, TypeGuard, cast
 
 _POS, _NEG = "\x00nnnotes:+inf", "\x00nnnotes:-inf"  # placeholders, replaced after encoding
 _TOKENS = ((json.dumps(_POS), "1e999"), (json.dumps(_NEG), "-1e999"))
 _MARK = json.dumps(_POS)[1 : -len("+inf") - 1]  # the placeholders' common prefix as JSON text
 
 
+def is_object(v: object) -> TypeGuard[dict[str, Any]]:
+    """Whether a decoded JSON value is an object."""
+    return isinstance(v, dict)
+
+
+def is_array(v: object) -> TypeGuard[list[Any]]:
+    """Whether a decoded JSON value is an array."""
+    return isinstance(v, list)
+
+
 class _NaN(ValueError):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__()
         self.path: list[str] = []
 
 
-def _finite(v, found: list):
+def _finite(v: object, found: list[bool]) -> object:
     """`v` with every non-finite float replaced by a placeholder string; containers are copied only
     where something changed, so the caller's document is never modified."""
     if isinstance(v, float):
@@ -38,8 +51,9 @@ def _finite(v, found: list):
         found.append(True)
         return _POS if v > 0 else _NEG
     if isinstance(v, dict):
-        out = None
-        for k, x in v.items():
+        d = cast("dict[object, object]", v)
+        out: dict[object, object] | None = None
+        for k, x in d.items():
             if k == _POS or k == _NEG:
                 raise ValueError(f"key {k!r} collides with the non-finite placeholder")
             try:
@@ -49,28 +63,29 @@ def _finite(v, found: list):
                 raise
             if y is not x:
                 if out is None:
-                    out = dict(v)
+                    out = dict(d)
                 out[k] = y
-        return v if out is None else out
+        return d if out is None else out
     if isinstance(v, (list, tuple)):
-        out = None
-        for i, x in enumerate(v):
+        items = cast("Sequence[object]", v)
+        copied: list[object] | None = None
+        for i, x in enumerate(items):
             try:
                 y = _finite(x, found)
             except _NaN as e:
                 e.path.append(f"[{i}]")
                 raise
             if y is not x:
-                if out is None:
-                    out = list(v)
-                out[i] = y
-        return v if out is None else out
+                if copied is None:
+                    copied = list(items)
+                copied[i] = y
+        return items if copied is None else copied
     if isinstance(v, str) and (v == _POS or v == _NEG):
         raise ValueError(f"string {v!r} collides with the non-finite placeholder")
     return v
 
 
-def dumps(obj, **kw) -> str:
+def dumps(obj: object, **kw: Any) -> str:
     """json.dumps with non-finite floats as 1e999 / -1e999; raises ValueError on NaN (with its path).
     Keyword arguments are json.dumps' (indent, ensure_ascii, separators, default, ...)."""
     try:
@@ -80,7 +95,7 @@ def dumps(obj, **kw) -> str:
     else:
         if _MARK not in s:
             return s  # (a placeholder string in the data: the walk rejects it)
-    found: list = []
+    found: list[bool] = []
     try:
         clean = _finite(obj, found)
     except _NaN as e:
@@ -92,7 +107,9 @@ def dumps(obj, **kw) -> str:
     return s
 
 
-def write_json(path, obj, *, indent: int | None = 1, ensure_ascii: bool = False, **kw) -> Path:
+def write_json(
+    path: str | os.PathLike[str], obj: object, *, indent: int | None = 1, ensure_ascii: bool = False, **kw: Any
+) -> Path:
     """Write `obj` as JSON to `path` (UTF-8, LF); see `dumps`. Returns the path."""
     path = Path(path)
     path.write_text(dumps(obj, indent=indent, ensure_ascii=ensure_ascii, **kw), encoding="utf-8", newline="\n")

src/nnnotes/schemas/__init__.py

@@ -3,6 +3,7 @@
 
 import json
 from importlib import resources
+from typing import Any
 
 __all__ = ["load", "names"]
 
@@ -15,7 +16,7 @@ def names() -> list[str]:
     return sorted(p.name[: -len(SUFFIX)] for p in files.iterdir() if p.name.endswith(SUFFIX))
 
 
-def load(name: str) -> dict:
+def load(name: str) -> dict[str, Any]:
     """The schema `name` (`music-data` for music-data.schema.json)."""
     path = resources.files("nnnotes.schemas") / f"{name}{SUFFIX}"
     if not path.is_file():