fix: Add type annotations, fix a bug in loading schema path (#2264)

* type: Annotate bst.utils

* type: Annotate types.namespace

* type: Annotate bst.schema

* chore: Update pre-commit hooks

Author: effigies

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
       - id: check-added-large-files
       - id: check-case-conflict
   - repo: https://github.com/python-jsonschema/check-jsonschema
-    rev: 0.33.3
+    rev: 0.35.0
     hooks:
       - id: check-dependabot
       - id: check-github-workflows
@@ -49,7 +49,7 @@ repos:
       - id: codespell
         args: ["--config=.codespellrc", "--dictionary=-", "--dictionary=.codespell_dict"]
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.17.1
+    rev: v1.18.2
     hooks:
       - id: mypy
         # Sync with project.optional-dependencies.typing

tools/schemacode/src/bidsschematools/render/tsv.py

@@ -12,7 +12,7 @@
 
 
 @propagate_fence_exception
-@in_context(WarningsFilter(["error"]))
+@in_context(WarningsFilter(("error",)))
 def fence(
     source: str,
     language: str,

tools/schemacode/src/bidsschematools/schema.py

@@ -3,9 +3,8 @@
 from __future__ import annotations
 
 import json
-import os
 import re
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Mapping, MutableMapping
 from copy import deepcopy
 from functools import cache, lru_cache
 from pathlib import Path
@@ -15,6 +14,9 @@
 
 TYPE_CHECKING = False
 if TYPE_CHECKING:
+    from typing import Any, Callable
+
+    from acres import typ as at
     from jsonschema.protocols import Validator as JsonschemaValidator
 
 lgr = utils.get_logger()
@@ -24,39 +26,38 @@ class BIDSSchemaError(Exception):
     """Errors indicating invalid values in the schema itself"""
 
 
-def _get_schema_version(schema_dir):
+def _get_schema_version(schema_dir: str | at.Traversable) -> str:
     """
     Determine schema version for given schema directory, based on file specification.
     """
+    if isinstance(schema_dir, str):
+        schema_dir = Path(schema_dir)
 
-    schema_version_path = os.path.join(schema_dir, "SCHEMA_VERSION")
-    with open(schema_version_path) as f:
-        schema_version = f.readline().rstrip()
-    return schema_version
+    schema_version_path = schema_dir / "SCHEMA_VERSION"
+    return schema_version_path.read_text().strip()
 
 
-def _get_bids_version(schema_dir):
+def _get_bids_version(schema_dir: str | at.Traversable) -> str:
     """
     Determine BIDS version for given schema directory, with directory name, file specification,
     and string fallback.
     """
+    if isinstance(schema_dir, str):
+        schema_dir = Path(schema_dir)
 
-    bids_version_path = os.path.join(schema_dir, "BIDS_VERSION")
+    bids_version_path = schema_dir / "BIDS_VERSION"
     try:
-        with open(bids_version_path) as f:
-            bids_version = f.readline().rstrip()
+        return bids_version_path.read_text().strip()
     # If this file is not in the schema, fall back to placeholder heuristics:
     except FileNotFoundError:
         # Maybe the directory encodes the version, as in:
         # https://github.com/bids-standard/bids-schema
-        _, bids_version = os.path.split(schema_dir)
-        if not re.match(r"^.*?[0-9]*?\.[0-9]*?\.[0-9]*?.*?$", bids_version):
-            # Then we don't know, really.
-            bids_version = schema_dir
-    return bids_version
+        if re.match(r"^.*?[0-9]*?\.[0-9]*?\.[0-9]*?.*?$", schema_dir.name):
+            return schema_dir.name
+    return str(schema_dir)
 
 
-def _find(obj, predicate):
+def _find(obj: object, predicate: Callable[[Any], bool]) -> Iterable[object]:
     """Find objects in an arbitrary object that satisfy a predicate.
 
     Note that this does not cut branches, so every iterable sub-object
@@ -78,7 +79,7 @@ def _find(obj, predicate):
     except Exception:
         pass
 
-    iterable = ()
+    iterable: Iterable[object] = ()
     if isinstance(obj, Mapping):
         iterable = obj.values()
     elif not isinstance(obj, str) and isinstance(obj, Iterable):
@@ -88,16 +89,17 @@ def _find(obj, predicate):
         yield from _find(item, predicate)
 
 
-def _dereference(namespace, base_schema):
+def _dereference(namespace: MutableMapping, base_schema: Namespace) -> None:
     # In-place, recursively dereference objects
     # This allows a referenced object to itself contain a reference
     # A dependency graph could be constructed, but would likely be slower
     # to build than to duplicate a couple dereferences
     for struct in _find(namespace, lambda obj: "$ref" in obj):
+        assert isinstance(struct, MutableMapping)
         target = base_schema.get(struct["$ref"])
         if target is None:
             raise ValueError(f"Reference {struct['$ref']} not found in schema.")
-        if isinstance(target, Mapping):
+        if isinstance(target, MutableMapping):
             struct.pop("$ref")
             _dereference(target, base_schema)
             struct.update({**target, **struct})
@@ -110,7 +112,7 @@ def get_schema_validator() -> JsonschemaValidator:
     return utils.jsonschema_validator(metaschema, check_format=True)
 
 
-def dereference(namespace, inplace=True):
+def dereference(namespace: Namespace, inplace: bool = True) -> Namespace:
     """Replace references in namespace with the contents of the referred object.
 
     Parameters
@@ -133,6 +135,8 @@ def dereference(namespace, inplace=True):
 
     # At this point, any remaining refs are one-off objects in lists
     for struct in _find(namespace, lambda obj: any("$ref" in sub for sub in obj)):
+        assert isinstance(struct, list)
+        item: MutableMapping[str, object]
         for i, item in enumerate(struct):
             try:
                 target = item.pop("$ref")
@@ -144,7 +148,7 @@ def dereference(namespace, inplace=True):
     return namespace
 
 
-def flatten_enums(namespace, inplace=True):
+def flatten_enums(namespace: Namespace, inplace=True) -> Namespace:
     """Replace enum collections with a single enum, merging enums contents.
 
     The function helps reducing the complexity of the schema by assuming
@@ -175,6 +179,7 @@ def flatten_enums(namespace, inplace=True):
     if not inplace:
         namespace = deepcopy(namespace)
     for struct in _find(namespace, lambda obj: "anyOf" in obj):
+        assert isinstance(struct, MutableMapping)
         try:
             # Deduplicate because JSON schema validators may not like duplicates
             # Long run, we should get rid of this function and have the rendering
@@ -189,7 +194,7 @@ def flatten_enums(namespace, inplace=True):
 
 
 @lru_cache
-def load_schema(schema_path=None):
+def load_schema(schema_path: at.Traversable | str | None = None) -> Namespace:
     """Load the schema into a dict-like structure.
 
     This function allows the schema, like BIDS itself, to be specified in
@@ -221,6 +226,7 @@ def load_schema(schema_path=None):
 
             # Probably a Windows checkout with a git link. Resolve first.
             if schema_path.is_file() and (content := schema_path.read_text()).startswith("../"):
+                assert isinstance(schema_path, Path)
                 schema_path = Path.resolve(schema_path.parent / content)
         lgr.info("No schema path specified, defaulting to the bundled schema, `%s`.", schema_path)
     elif isinstance(schema_path, str):

tools/schemacode/src/bidsschematools/types/_generator.py

@@ -18,6 +18,7 @@
 
 TYPE_CHECKING = False
 if TYPE_CHECKING:
+    from collections.abc import Mapping
     from typing import Any, Callable, Protocol
 
     class Spec(Protocol):
@@ -241,7 +242,7 @@ def generate_protocols(
     return protocols
 
 
-def generate_module(schema: dict[str, Any], class_type: str) -> str:
+def generate_module(schema: Mapping[str, Any], class_type: str) -> str:
     """Generate a context module source code from a BIDS schema.
 
     Returns a tuple containing the module source code and a list of protocol names.

tools/schemacode/src/bidsschematools/types/namespace.py

@@ -4,14 +4,22 @@
 YAML files available as a single dictionary and allow attribute (``.``)
 lookups.
 """
+from __future__ import annotations
 
 import json
-import typing as ty
+import os.path
 from collections.abc import ItemsView, KeysView, Mapping, MutableMapping, ValuesView
 from pathlib import Path
 
+TYPE_CHECKING = False
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from typing import Any, Self
 
-def _expand_dots(entry: tuple[str, ty.Any]) -> tuple[str, ty.Any]:
+    from acres import typ as at
+
+
+def _expand_dots(entry: tuple[str, Any]) -> tuple[str, Any]:
     # Helper function for expand
     key, val = entry
     if "." in key:
@@ -20,7 +28,7 @@ def _expand_dots(entry: tuple[str, ty.Any]) -> tuple[str, ty.Any]:
     return key, expand(val)
 
 
-def expand(element):
+def expand(element: dict[str, Any]) -> dict[str, Any]:
     """Expand a dict, recursively, to replace dots in keys with recursive dictionaries
 
     Parameters
@@ -46,18 +54,18 @@ def expand(element):
 
 
 class NsItemsView(ItemsView):
-    def __init__(self, namespace, level):
+    def __init__(self, namespace: Mapping, level: int):
         self._mapping = namespace
         self._level = level
 
-    def __contains__(self, item):
+    def __contains__(self, item: Any) -> bool:
         key, val = item
         keys = key.split(".", self._level - 1)
         if "." in keys[-1]:
             return False
         return self._mapping[key] == val
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[tuple[str, Any]]:
         l1 = ItemsView(self._mapping)
         if self._level == 1:
             yield from l1
@@ -73,30 +81,30 @@ def __iter__(self):
 
 
 class NsKeysView(KeysView):
-    def __init__(self, namespace, level):
+    def __init__(self, namespace: Mapping, level: int):
         self._mapping = namespace
         self._level = level
 
-    def __contains__(self, key):
+    def __contains__(self, key: Any) -> bool:
         keys = key.split(".", self._level - 1)
         if "." in keys[-1]:
             return False
         return key in self._mapping
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[str]:
         yield from (key for key, val in NsItemsView(self._mapping, self._level))
 
 
 class NsValuesView(ValuesView):
-    def __init__(self, namespace, level):
+    def __init__(self, namespace: Mapping, level: int):
         self._mapping = namespace
         self._level = level
         self._items = NsItemsView(namespace, level)
 
-    def __contains__(self, val):
+    def __contains__(self, val: object) -> bool:
         return any(val == item[1] for item in self._items)
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[Any]:
         yield from (val for key, val in self._items)
 
 
@@ -162,7 +170,7 @@ class Namespace(MutableMapping):
     >>> del ns['d']
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         self._properties = dict(*args, **kwargs)
 
     def to_dict(self) -> dict:
@@ -177,7 +185,7 @@ def _to_dict(obj):
 
         return _to_dict(self)
 
-    def __deepcopy__(self, memo):
+    def __deepcopy__(self, memo) -> Self:
         return self.build(self.to_dict())
 
     @classmethod
@@ -218,42 +226,43 @@ def __getattribute__(self, key):
         except KeyError:
             raise err
 
-    def _get_mapping(self, key: str) -> tuple[Mapping, str]:
+    def _get_mapping(self, key: str) -> tuple[MutableMapping, str]:
         subkeys = key.split(".")
         mapping = self._properties
         for subkey in subkeys[:-1]:
             mapping = mapping.setdefault(subkey, {})
-            mapping = getattr(mapping, "_properties", mapping)
+            if isinstance(mapping, Namespace):
+                mapping = mapping._properties
             if not isinstance(mapping, Mapping):
                 raise KeyError(f"{key} (subkey={subkey})")
         return mapping, subkeys[-1]
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: str) -> Any:
         mapping, subkey = self._get_mapping(key)
         val = mapping[subkey]
         if isinstance(val, dict):
             val = self.view(val)
         return val
 
-    def __setitem__(self, key, val):
+    def __setitem__(self, key: str, val: Any):
         mapping, subkey = self._get_mapping(key)
         mapping[subkey] = val
 
-    def __delitem__(self, key):
+    def __delitem__(self, key: str):
         mapping, subkey = self._get_mapping(key)
         del mapping[subkey]
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<Namespace {self._properties}>"
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self._properties)
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[str]:
         return iter(self._properties)
 
     @classmethod
-    def from_directory(cls, path, fmt="yaml"):
+    def from_directory(cls, path: at.Traversable | str, fmt: str = "yaml") -> Self:
         if fmt == "yaml":
             if isinstance(path, str):
                 path = Path(path)
@@ -264,27 +273,27 @@ def to_json(self, **kwargs) -> str:
         return json.dumps(self, cls=MappingEncoder, **kwargs)
 
     @classmethod
-    def from_json(cls, jsonstr: str):
+    def from_json(cls, jsonstr: str) -> Self:
         return cls.build(json.loads(jsonstr))
 
 
-def _read_yaml_dir(path: Path) -> dict:
+def _read_yaml_dir(path: at.Traversable) -> dict:
     mapping = {}
-    for subpath in sorted(path.iterdir()):
+    for subpath in sorted(path.iterdir(), key=lambda p: p.name):
         if subpath.is_dir():
             mapping[subpath.name] = _read_yaml_dir(subpath)
         elif subpath.name.endswith("yaml"):
             import yaml
 
             try:
-                mapping[subpath.stem] = yaml.safe_load(subpath.read_text())
+                mapping[os.path.splitext(subpath.name)[0]] = yaml.safe_load(subpath.read_text())
             except Exception as e:
                 raise ValueError(f"There was an error reading the file: {subpath}") from e
     return mapping
 
 
 class MappingEncoder(json.JSONEncoder):
-    def default(self, o):
+    def default(self, o: object) -> object:
         try:
             return super().default(o)
         except TypeError as e: