[NEAT-243]😮‍💨Cleanup issues part 6

cognite/neat/issues/_base.py

@@ -67,7 +67,7 @@ class NeatError(NeatIssue, ABC):
     def dump(self) -> dict[str, Any]:
         return {"errorType": type(self).__name__}
 
-    def as_exception(self) -> ValueError:
+    def as_exception(self) -> Exception:
         return ValueError(self.message())
 
     def as_pydantic_exception(self) -> PydanticCustomError:
@@ -85,10 +85,11 @@ def from_pydantic_errors(cls, errors: list[ErrorDetails], **kwargs) -> "list[Nea
         """
         all_errors: list[NeatError] = []
         for error in errors:
-            if isinstance(ctx := error.get("ctx"), dict) and isinstance(
-                multi_error := ctx.get("error"), MultiValueError
-            ):
+            ctx = error.get("ctx")
+            if isinstance(ctx, dict) and isinstance(multi_error := ctx.get("error"), MultiValueError):
                 all_errors.extend(multi_error.errors)  # type: ignore[arg-type]
+            elif isinstance(ctx, dict) and isinstance(single_error := ctx.get("error"), NeatError):
+                all_errors.append(single_error)
             else:
                 all_errors.append(DefaultPydanticError.from_pydantic_error(error))
         return all_errors

cognite/neat/issues/errors/external.py

@@ -2,7 +2,10 @@
 from pathlib import Path
 from typing import Any
 
+from yaml import YAMLError
+
 from cognite.neat.issues import NeatError
+from cognite.neat.utils.text import humanize_sequence
 
 
 @dataclass(frozen=True)
@@ -47,6 +50,9 @@ class NeatFileNotFoundError(NeatError):
     fix = "Make sure to provide a valid file"
     filepath: Path
 
+    def as_exception(self) -> Exception:
+        return FileNotFoundError(self.message())
+
     def message(self) -> str:
         return (__doc__ or "").format(filepath=repr(self.filepath))
 
@@ -73,3 +79,70 @@ def dump(self) -> dict[str, Any]:
         output["filepath"] = self.filepath
         output["field_type"] = self.field_name
         return output
+
+
+@dataclass(frozen=True)
+class InvalidYamlError(NeatError):
+    """Invalid YAML: {reason}"""
+
+    extra = "Expected format: {expected_format}"
+    fix = "Check if the file is a valid YAML file"
+
+    reason: str
+    expected_format: str | None = None
+
+    def as_exception(self) -> YAMLError:
+        return YAMLError(self.message())
+
+    def message(self) -> str:
+        msg = (self.__doc__ or "").format(reason=self.reason)
+        if self.expected_format:
+            msg += f" {self.extra.format(expected_format=self.expected_format)}"
+        return msg
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["reason"] = self.reason
+        output["expected_format"] = self.expected_format
+        return output
+
+
+@dataclass(frozen=True)
+class UnexpectedFileTypeError(NeatError):
+    """Unexpected file type: {filepath}. Expected format: {expected_format}"""
+
+    filepath: Path
+    expected_format: list[str]
+
+    def as_exception(self) -> Exception:
+        return TypeError(self.message())
+
+    def message(self) -> str:
+        return (__doc__ or "").format(
+            filepath=repr(self.filepath), expected_format=humanize_sequence(self.expected_format)
+        )
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["filepath"] = self.filepath
+        output["expected_format"] = self.expected_format
+        return output
+
+
+@dataclass(frozen=True)
+class FileNotAFileError(NeatError):
+    """{filepath} is not a file"""
+
+    fix = "Make sure to provide a valid file"
+    filepath: Path
+
+    def as_exception(self) -> Exception:
+        return FileNotFoundError(self.message())
+
+    def message(self) -> str:
+        return (__doc__ or "").format(filepath=repr(self.filepath))
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["filepath"] = self.filepath
+        return output

cognite/neat/issues/neat_warnings/external.py

@@ -0,0 +1,84 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from cognite.neat.issues import NeatWarning
+from cognite.neat.utils.text import humanize_sequence
+
+
+@dataclass(frozen=True)
+class FileReadWarning(NeatWarning):
+    """Error when reading file, {filepath}: {reason}"""
+
+    filepath: Path
+    reason: str
+
+    def message(self) -> str:
+        return (self.__doc__ or "").format(filepath=repr(self.filepath), reason=self.reason)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["filepath"] = self.filepath
+        output["reason"] = self.reason
+        return output
+
+
+@dataclass(frozen=True)
+class FileMissingRequiredFieldWarning(NeatWarning):
+    """Missing required {field_name} in {filepath}: {field}. The file will be skipped"""
+
+    filepath: Path
+    field_name: str
+    field: str
+
+    def message(self) -> str:
+        return (self.__doc__ or "").format(field_name=self.field_name, field=self.field, filepath=self.filepath)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["field_name"] = self.field_name
+        output["field"] = self.field
+        output["filepath"] = self.filepath
+        return output
+
+
+@dataclass(frozen=True)
+class UnexpectedFileTypeWarning(NeatWarning):
+    """Unexpected file type: {filepath}. Expected format: {expected_format}"""
+
+    extra = "Error: {error_message}"
+
+    filepath: Path
+    expected_format: list[str]
+    error_message: str | None = None
+
+    def message(self) -> str:
+        msg = (__doc__ or "").format(
+            filepath=repr(self.filepath), expected_format=humanize_sequence(self.expected_format)
+        )
+        if self.error_message:
+            msg += f" {self.extra.format(error_message=self.error_message)}"
+        return msg
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["filepath"] = self.filepath
+        output["expected_format"] = self.expected_format
+        return output
+
+
+@dataclass(frozen=True)
+class UnknownItemWarning(NeatWarning):
+    """Unknown item {item} in {filepath}. The item will be skipped"""
+
+    item: str
+    filepath: Path
+
+    def message(self) -> str:
+        return (self.__doc__ or "").format(item=self.item, filepath=self.filepath)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["item"] = self.item
+        output["filepath"] = self.filepath
+        return output

cognite/neat/issues/neat_warnings/general.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+from typing import Any
+
+from cognite.neat.issues import NeatWarning
+
+
+@dataclass(frozen=True)
+class NeatValueWarning(NeatWarning):
+    """{value}"""
+
+    value: str
+
+    def message(self) -> str:
+        return (self.__doc__ or "").format(value=self.value)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["value"] = self.value
+        return output
+
+
+@dataclass(frozen=True)
+class NotSupportedWarning(NeatWarning):
+    """{feature} is not supported"""
+
+    feature: str
+
+    def message(self) -> str:
+        return (self.__doc__ or "").format(feature=self.feature)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["feature"] = self.feature
+        return output

cognite/neat/issues/neat_warnings/identifier.py

@@ -8,15 +8,20 @@
 class RegexViolationWarning(NeatWarning):
     """The value '{value}' of {identifier} does not match the {pattern_name} pattern '{pattern}'"""
 
+    extra = "{motivation}"
     value: str
     pattern: str
     identifier: str
     pattern_name: str
+    motivation: str | None = None
 
     def message(self) -> str:
-        return (self.__doc__ or "").format(
+        msg = (self.__doc__ or "").format(
             value=self.value, pattern=self.pattern, identifier=self.identifier, pattern_name=self.pattern_name
         )
+        if self.motivation:
+            msg += f"\n{self.extra.format(motivation=self.motivation)}"
+        return msg
 
     def dump(self) -> dict[str, Any]:
         output = super().dump()

cognite/neat/issues/neat_warnings/models.py

@@ -1,8 +1,28 @@
+import sys
 from dataclasses import dataclass
 from typing import Any
 
 from cognite.neat.issues import NeatWarning
 
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from backports.strenum import StrEnum
+
+_BASE_URL = "https://cognite-neat.readthedocs-hosted.com/en/latest/data-modeling-principles.html"
+
+
+class DataModelingPrinciple(StrEnum):
+    """Data modeling principles that are violated by a class."""
+
+    ONE_MODEL_ONE_SPACE = "all-data-models-are-kept-in-its-own-space"
+    SAME_VERSION = "all-views-of-a-data-models-have-the-same-version-and-space-as-the-data-model"
+    SOLUTION_BUILDS_ON_ENTERPRISE = "solution-data-models-should-always-be-referencing-the-enterprise-data-model"
+
+    @property
+    def url(self) -> str:
+        return f"{_BASE_URL}#{self.value}"
+
 
 @dataclass(frozen=True)
 class InvalidClassWarning(NeatWarning):
@@ -20,3 +40,69 @@ def dump(self) -> dict[str, Any]:
         output["class_name"] = self.class_name
         output["reason"] = self.reason
         return output
+
+
+@dataclass(frozen=True)
+class BreakingModelingPrincipleWarning(NeatWarning):
+    """{specific} violates the {principle} principle. See {url} for more information."""
+
+    specific: str
+    principle: DataModelingPrinciple
+
+    def message(self) -> str:
+        principle = self.principle.value.replace("_", " ").title()
+        return (self.__doc__ or "").format(specific=self.specific, principle=principle, url=self.principle.url)
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["specific"] = self.specific
+        output["principle"] = self.principle
+        return output
+
+
+@dataclass(frozen=True)
+class UserModelingWarning(NeatWarning):
+    """{title}: {problem}. {explanation}"""
+
+    extra = "Suggestion: {suggestion}"
+    title: str
+    problem: str
+    explanation: str
+    suggestion: str | None = None
+
+    def message(self) -> str:
+        msg = (self.__doc__ or "").format(title=self.title, problem=self.problem, explanation=self.explanation)
+        if self.suggestion:
+            msg += f"\n{self.extra.format(suggestion=self.suggestion)}"
+        return msg
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["title"] = self.title
+        output["problem"] = self.problem
+        output["explanation"] = self.explanation
+        output["suggestion"] = self.suggestion
+        return output
+
+
+@dataclass(frozen=True)
+class CDFNotSupportedWarning(NeatWarning):
+    """{title} - Will likely fail to write to CDF. {problem}."""
+
+    extra = "Suggestion: {suggestion}"
+    title: str
+    problem: str
+    suggestion: str | None = None
+
+    def message(self) -> str:
+        msg = (self.__doc__ or "").format(title=self.title, problem=self.problem)
+        if self.suggestion:
+            msg += f"\n{self.extra.format(suggestion=self.suggestion)}"
+        return msg
+
+    def dump(self) -> dict[str, Any]:
+        output = super().dump()
+        output["title"] = self.title
+        output["problem"] = self.problem
+        output["suggestion"] = self.suggestion
+        return output

cognite/neat/rules/exporters/_rules2dms.py

@@ -16,8 +16,8 @@
 from cognite.client.exceptions import CogniteAPIError
 
 from cognite.neat.issues import IssueList
+from cognite.neat.issues.neat_warnings.models import BreakingModelingPrincipleWarning, DataModelingPrinciple
 from cognite.neat.issues.neat_warnings.resources import FailedLoadingResourcesWarning
-from cognite.neat.rules import issues
 from cognite.neat.rules._shared import Rules
 from cognite.neat.rules.models import InformationRules
 from cognite.neat.rules.models.dms import DMSRules, DMSSchema, PipelineSchema
@@ -298,7 +298,11 @@ def _validate(self, loader: ResourceLoader, items: CogniteResourceList) -> Issue
         if isinstance(loader, DataModelLoader):
             models = cast(list[DataModelApply], items)
             if other_models := self._exist_other_data_models(loader, models):
-                warning = issues.dms.OtherDataModelsInSpaceWarning(models[0].space, other_models)
+                warning = BreakingModelingPrincipleWarning(
+                    f"There are multiple data models in the same space {models[0].space}. "
+                    f"Other data models in the space are {other_models}.",
+                    DataModelingPrinciple.ONE_MODEL_ONE_SPACE,
+                )
                 if not self.suppress_warnings:
                     warnings.warn(warning, stacklevel=2)
                 issue_list.append(warning)