from typing import List
-from knot_resolver import KresBaseException
+from knot_resolver import KresBaseError
-class SubprocessControllerError(KresBaseException):
- pass
+class KresSubprocessControllerError(KresBaseError):
+ """Class for errors that are raised in the controller module."""
-class SubprocessControllerExecError(Exception):
+class KresSubprocessControllerExec(Exception): # noqa: N818
"""
- Exception that is used to deliberately terminate system startup
- and make exec() of something else. This is used by the subprocess controller
- as supervisord to run as the top-level process in a process tree hierarchy.
+ Custom non-error exception that indicates the need for exec().
+
+ Raised by the controller (supervisord) and caught by the controlled process (manager).
+ The exception says that the process needs to perform a re-exec during startup.
+ This ensures that the process runs under the controller (supervisord) in a process tree hierarchy.
"""
def __init__(self, exec_args: List[str], *args: object) -> None:
-from typing import Iterable, List
+from typing import Iterable, Iterator
-from knot_resolver import KresBaseException
+from knot_resolver import KresBaseError
-class DataModelingBaseException(KresBaseException):
- """
- Base class for all exceptions used in modelling.
- """
+class ModelingBaseError(KresBaseError):
+ """Base class for all errors used in data modeling."""
-class DataParsingError(DataModelingBaseException):
- pass
+class DataDescriptionError(ModelingBaseError):
+ """Class for errors that are raised when checking data description."""
-class DataDescriptionError(DataModelingBaseException):
- pass
+class DataParsingError(ModelingBaseError):
+ """Class for errors that are raised when parsing data."""
-class DataValidationError(DataModelingBaseException):
- def __init__(self, msg: str, tree_path: str, child_exceptions: "Iterable[DataValidationError]" = tuple()) -> None:
+class DataValidationError(ModelingBaseError):
+ """Class for errors that are raised when validating data."""
+
+ def __init__(self, msg: str, tree_path: str, child_exceptions: Iterable["DataValidationError"] = ()) -> None:
super().__init__(msg)
self._tree_path = tree_path.replace("_", "-")
self._child_exceptions = child_exceptions
def where(self) -> str:
return self._tree_path
- def msg(self):
+ def msg(self) -> str:
return f"[{self.where()}] {super().__str__()}"
def recursive_msg(self, indentation_level: int = 0) -> str:
- msg_parts: List[str] = []
+ def indented_lines(level: int) -> Iterator[str]:
+ if level == 0:
+ yield "Configuration validation error detected:"
+ level += 1
- if indentation_level == 0:
- indentation_level += 1
- msg_parts.append("Configuration validation error detected:")
+ indent = "\t" * level
+ yield f"{indent}{self.msg()}"
- indent = indentation_level * "\t"
- msg_parts.append(f"{indent}{self.msg()}")
+ for child in self._child_exceptions:
+ yield from child.recursive_msg(level + 1).split("\n")
- for c in self._child_exceptions:
- msg_parts.append(c.recursive_msg(indentation_level + 1))
- return "\n".join(msg_parts)
+ return "\n".join(indented_lines(indentation_level))
def __str__(self) -> str:
return self.recursive_msg()
class AggregateDataValidationError(DataValidationError):
- def __init__(self, object_path: str, child_exceptions: "Iterable[DataValidationError]") -> None:
+ """Aggregation class for errors (DataValidationError) raised during data validation."""
+
+ def __init__(self, object_path: str, child_exceptions: Iterable[DataValidationError]) -> None:
super().__init__("error due to lower level exceptions", object_path, child_exceptions)
def recursive_msg(self, indentation_level: int = 0) -> str:
- inc = 0
- msg_parts: List[str] = []
- if indentation_level == 0:
- inc = 1
- msg_parts.append("Configuration validation errors detected:")
-
- for c in self._child_exceptions:
- msg_parts.append(c.recursive_msg(indentation_level + inc))
- return "\n".join(msg_parts)
+ def indented_lines(level: int) -> Iterator[str]:
+ inc = 0
+ if level == 0:
+ yield "Configuration validation errors detected:"
+ inc = 1
+
+ for child in self._child_exceptions:
+ yield from child.recursive_msg(level + inc).split("\n")
+
+ return "\n".join(indented_lines(indentation_level))