adapt exceptions

pyproject.toml

@@ -17,7 +17,7 @@ namespaces = true
 # ----------------------------------------- Project Metadata -------------------------------------
 #
 [project]
-version = "0.0.0.dev267"
+version = "0.0.0.dev268"
 name = "ControlMan"
 dependencies = [
     "packaging >= 23.2, < 24",

src/controlman/__init__.py

@@ -12,6 +12,7 @@
 from controlman import const, exception
 from controlman import data_validator as _data_validator
 from controlman import _file_util
+from controlman import exception as _exception
 from controlman.center_manager import CenterManager
 
 
@@ -54,7 +55,6 @@ def manager(
 def from_json_file(
     repo_path: str | _Path,
     filepath: str = const.FILEPATH_METADATA,
-    validate: bool = True,
 ) -> _ps.NestedDict:
     """Load control center data from the full JSON file.
 
@@ -64,42 +64,40 @@ def from_json_file(
         Path to the repository root.
     filepath : str, default: controlman.const.FILEPATH_METADATA
         Relative path to the JSON file in the repository.
-    validate : bool, default: True
-        Validate the data against the schema.
 
     Raises
     ------
     controlman.exception.ControlManFileReadError
         If the file cannot be read.
     """
-    data_dict = _file_util.read_data_from_file(
-        path=_Path(repo_path) / filepath,
-        base_path=repo_path,
-        extension="json",
-        raise_errors=True,
-    )
-    if validate:
-        _data_validator.validate(data=data_dict)
-    return _ps.NestedDict(data_dict)
+    try:
+        data = _ps.read.json_from_file(path=_Path(repo_path) / filepath)
+    except _ps.exception.read.PySerialsReadException as e:
+        raise _exception.load.ControlManInvalidMetadataError(cause=e, filepath=filepath) from None
+    _data_validator.validate(data=data)
+    return _ps.NestedDict(data)
 
 
 def from_json_file_at_commit(
     git_manager: _Git,
     commit_hash: str,
     filepath: str = const.FILEPATH_METADATA,
-    validate: bool = True,
 ) -> _ps.NestedDict:
     data_str = git_manager.file_at_hash(
         commit_hash=commit_hash,
         path=filepath,
     )
-    return from_json_string(data=data_str, validate=validate)
+    try:
+        data = _ps.read.json_from_string(data=data_str)
+    except _ps.exception.read.PySerialsReadException as e:
+        raise _exception.load.ControlManInvalidMetadataError(
+            cause=e, filepath=filepath, commit_hash=commit_hash
+        ) from None
+    _data_validator.validate(data=data)
+    return _ps.NestedDict(data)
 
 
-def from_json_string(
-    data: str,
-    validate: bool = True,
-) -> _ps.NestedDict:
+def from_json_string(data: str) -> _ps.NestedDict:
     """Load control center data from the full JSON string.
 
     Parameters
@@ -114,11 +112,9 @@ def from_json_string(
     controlman.exception.ControlManFileReadError
         If the data cannot be read.
     """
-    data_dict = _file_util.read_datafile_from_string(
-        data=data,
-        extension="json",
-        raise_errors=True,
-    )
-    if validate:
-        _data_validator.validate(data=data_dict)
-    return _ps.NestedDict(data_dict)
+    try:
+        data = _ps.read.json_from_string(data=data)
+    except _ps.exception.read.PySerialsReadException as e:
+        raise _exception.load.ControlManInvalidMetadataError(e) from None
+    _data_validator.validate(data=data)
+    return _ps.NestedDict(data)

src/controlman/_file_util.py

@@ -1,74 +1,10 @@
-from pathlib import Path
-from typing import Literal
-
 import pkgdata as _pkgdata
-import pyserials
-
-from controlman import exception as _exception
+import pyserials as _ps
 
 
 _data_dir_path = _pkgdata.get_package_path_from_caller(top_level=True) / "_data"
 
 
-def read_data_from_file(
-    path: Path | str,
-    base_path: Path | str | None = None,
-    extension: Literal["json", "yaml", "toml"] | None = None,
-    raise_errors: bool = True,
-) -> dict | None:
-    try:
-        data = pyserials.read.from_file(
-            path=path,
-            data_type=extension,
-            json_strict=True,
-            yaml_safe=True,
-            toml_as_dict=False,
-        )
-    except pyserials.exception.read.PySerialsReadException as e:
-        if raise_errors:
-            raise _exception.ControlManFileReadError(
-                path=Path(path).relative_to(base_path) if base_path else path,
-                data=getattr(e, "data", None),
-            ) from e
-        return
-    if not isinstance(data, dict):
-        if raise_errors:
-            raise _exception.ControlManFileDataTypeError(
-                expected_type=dict,
-                path=Path(path).relative_to(base_path) if base_path else path,
-                data=data,
-            )
-        return
-    return data
-
-
-def read_datafile_from_string(
-    data: str,
-    extension: Literal["json", "yaml", "toml"],
-    raise_errors: bool = True,
-) -> dict | None:
-    try:
-        data = pyserials.read.from_string(
-            data=data,
-            data_type=extension,
-            json_strict=True,
-            yaml_safe=True,
-            toml_as_dict=False,
-        )
-    except pyserials.exception.read.PySerialsReadException as e:
-        if raise_errors:
-            raise _exception.ControlManFileReadError(data=data) from e
-        return
-    if not isinstance(data, dict):
-        if raise_errors:
-            raise _exception.ControlManFileDataTypeError(
-                expected_type=dict,
-                data=data,
-            )
-        return
-    return data
-
-
 def get_package_datafile(path: str) -> str | dict | list:
     """
     Get a data file in the package's '_data' directory.
@@ -81,5 +17,5 @@ def get_package_datafile(path: str) -> str | dict | list:
     full_path = _data_dir_path / path
     data = full_path.read_text()
     if full_path.suffix == ".yaml":
-        return pyserials.read.yaml_from_string(data=data, safe=True)
+        return _ps.read.yaml_from_string(data=data, safe=True)
     return data

src/controlman/cache_manager.py

@@ -3,7 +3,7 @@
 
 
 from loggerman import logger as _logger
-import pyserials as _pyserials
+import pyserials as _ps
 
 from controlman import exception as _exception, const as _const, _file_util
 from controlman import data_validator as _data_validator
@@ -26,13 +26,8 @@ def __init__(
             self._cache = {}
         else:
             try:
-                self._cache = _file_util.read_data_from_file(
-                    path=self._path,
-                    base_path=path_repo,
-                    extension="yaml",
-                    raise_errors=True
-                )
-            except _exception.ControlManException as e:
+                self._cache = _ps.read.yaml_from_file(path=self._path)
+            except _ps.exception.read.PySerialsReadException as e:
                 self._cache = {}
                 _logger.info(
                     "Caching", f"API cache file at '{self._path}' is corrupted; initialized new cache."
@@ -81,7 +76,7 @@ def set(self, typ: str, key: str, value: dict | list | str | int | float | bool)
         return
 
     def save(self):
-        _pyserials.write.to_yaml_file(
+        _ps.write.to_yaml_file(
             data=self._cache,
             path=self._path,
             make_dirs=True,

src/controlman/center_manager.py

@@ -70,7 +70,7 @@ def load(self) -> _ps.NestedDict:
                 const.FUNCNAME_CC_HOOK_POST_LOAD,
                 full_data,
             )
-        _data_validator.validate(data=full_data, before_substitution=True)
+        _data_validator.validate(data=full_data, source="source", before_substitution=True)
         self._data_raw = _ps.NestedDict(full_data)
         return self._data_raw
 
@@ -94,7 +94,7 @@ def generate_data(self) -> _ps.NestedDict:
             )
         self._cache_manager.save()
         data.fill()
-        _data_validator.validate(data=data())
+        _data_validator.validate(data=data(), source="source")
         self._data = data
         return self._data
 

src/controlman/data_gen/main.py

@@ -59,11 +59,11 @@ def _repo(self) -> None:
             fallback_purpose=False
         )
         if not repo_address:
-            _exception.ControlManRepositoryError(
-                "Failed to determine GitHub address. "
-                "The Git repository has not remote set for 'push' to 'origin'. "
-                f"Following remotes were found: {str(self._git.get_remotes())}",
+            raise _exception.data_gen.ControlManRepositoryError(
                 repo_path=self._git.repo_path,
+                description="Failed to determine GitHub address. "
+                "The Git repository has no remote set for push to origin. "
+                f"Following remotes were found: {str(self._git.get_remotes())}",
             )
         username, repo_name = repo_address
         self._gh_api_repo = self._gh_api.user(username).repo(repo_name)
@@ -127,9 +127,12 @@ def _license(self):
         if not license_info:
             for key in ("name", "text", "notice"):
                 if key not in data:
-                    raise _exception.ControlManSchemaValidationError(
-                        f"`license.{key}` is required when `license.id` is not a supported ID.",
-                        key="license"
+                    raise _exception.load.ControlManSchemaValidationError(
+                        source="source",
+                        before_substitution=True,
+                        description=f"`license.{key}` is required when `license.id` is not a supported ID.",
+                        json_path="license",
+                        data=self._data(),
                     )
             _logger.info("License data is manually set.")
             return
@@ -161,12 +164,14 @@ def _copyright(self):
             _logger.info(f"Project start year set from repository creation date: {start_year}")
         else:
             if start_year > current_year:
-                raise _exception.ControlManSchemaValidationError(
-                    msg=(
+                raise _exception.load.ControlManSchemaValidationError(
+                    source="source",
+                    description=(
                         f"Project start year ({start_year}) cannot be greater "
                         f"than current year ({current_year})."
                     ),
-                    key="copyright.start_year"
+                    json_path="copyright.start_year",
+                    data=self._data(),
                 )
             _logger.info(f"Project start year already set manually in metadata: {start_year}")
         year_range = f"{start_year}{'' if start_year == current_year else f'–{current_year}'}"

src/controlman/data_gen/python.py

@@ -74,17 +74,23 @@ def get_python_releases():
         version_spec_key = "pkg.python.version.spec"
         spec_str = self._data.fill(version_spec_key)
         if not spec_str:
-            _exception.ControlManSchemaValidationError(
-                "The package has not specified a Python version specifier.",
-                key=version_spec_key,
+            _exception.load.ControlManSchemaValidationError(
+                source="source",
+                before_substitution=True,
+                description="The package has not specified a Python version specifier.",
+                json_path=version_spec_key,
+                data=self._data(),
             )
         try:
             spec = _specifiers.SpecifierSet(spec_str)
         except _specifiers.InvalidSpecifier as e:
-            raise _exception.ControlManSchemaValidationError(
-                f"Invalid Python version specifier '{spec_str}'.",
-                key=version_spec_key,
-            ) from e
+            raise _exception.load.ControlManSchemaValidationError(
+                source="source",
+                before_substitution=True,
+                description=f"Invalid Python version specifier '{spec_str}'.",
+                json_path=version_spec_key,
+                data=self._data(),
+            ) from None
 
         current_python_versions = get_python_releases()
         micro_str = []
@@ -104,10 +110,13 @@ def get_python_releases():
             minor_str_pyxy.append(f"py{''.join(map(str, compat_ver_micro_int[:2]))}")
 
         if len(micro_str) == 0:
-            raise _exception.ControlManSchemaValidationError(
-                f"The Python version specifier '{spec_str}' does not match any "
+            raise _exception.load.ControlManSchemaValidationError(
+                source="source",
+                before_substitution=True,
+                description=f"The Python version specifier '{spec_str}' does not match any "
                 f"released Python version: '{current_python_versions}'.",
-                key=version_spec_key,
+                json_path=version_spec_key,
+                data=self._data(),
             )
         output = {
             "micros": sorted(micro_str, key=lambda x: tuple(map(int, x.split(".")))),
@@ -124,9 +133,12 @@ def get_python_releases():
     def _package_operating_systems(self):
         data_os = self._data.fill("pkg.os")
         if not isinstance(data_os, dict):
-            raise _exception.ControlManSchemaValidationError(
-                "The package has not specified any operating systems.",
-                key="pkg.os",
+            raise _exception.load.ControlManSchemaValidationError(
+                source="source",
+                before_substitution=True,
+                description="The package has not specified any operating systems.",
+                json_path="pkg.os",
+                data=self._data(),
             )
         pure_python = not any("ci_build" in os for os in data_os.values())
         self._data["pkg.python.pure"] = pure_python
@@ -182,9 +194,12 @@ def operating_system():
             classifiers = self._data.get(f"{path}.classifiers", [])
             for classifier in classifiers:
                 if classifier not in _trove_classifiers.classifiers:
-                    raise _exception.ControlManSchemaValidationError(
-                        f"Trove classifier '{classifier}' is not valid.",
-                        key=f"{path}.classifiers"
+                    raise _exception.load.ControlManSchemaValidationError(
+                        source="source",
+                        before_substitution=True,
+                        description=f"Trove classifier '{classifier}' is not valid.",
+                        json_path=f"{path}.classifiers",
+                        data=self._data(),
                     )
             classifiers.extend(common_classifiers)
             self._data[f"{path}.classifiers"] = sorted(classifiers)

src/controlman/data_gen/web.py

@@ -58,7 +58,7 @@ def _process_website_toctrees(self) -> None:
                     key_singular = key.removesuffix('s')
                     final_key = f"blog_{key_singular}_{value_slug}"
                     if final_key in pages:
-                        raise _exception.ControlManWebsiteError(
+                        raise _exception.data_gen.ControlManWebsiteError(
                             "Duplicate page ID. "
                             f"Generated ID '{final_key}' already exists "
                             f"for page '{pages[final_key]['path']}'. "