Skip to content

Base

cognite.neat.issues #

This is module contains all the Neat Exceptions (Errors) and Warnings as well as some helper classes to handle them like NeatIssueList

DefaultWarning dataclass #

Bases: NeatWarning

{category}: {warning}

Source code in cognite/neat/issues/_base.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
@dataclass(frozen=True)
class DefaultWarning(NeatWarning):
    """{category}: {warning}"""

    extra = "Source: {source}"

    warning: str
    category: str
    source: str | None = None

    @classmethod
    def from_warning_message(cls, warning: WarningMessage) -> NeatWarning:
        if isinstance(warning.message, NeatWarning):
            return warning.message

        return cls(
            warning=str(warning.message),
            category=warning.category.__name__,
            source=warning.source,
        )

    def as_message(self) -> str:
        return str(self.warning)

IssueList #

Bases: NeatIssueList[NeatIssue]

This is a list of NeatIssues.

Source code in cognite/neat/issues/_base.py 
402
403
404
405
class IssueList(NeatIssueList[NeatIssue]):
    """This is a list of NeatIssues."""

    ...

MultiValueError #

Bases: ValueError

This is a container for multiple errors.

It is used in the pydantic field_validator/model_validator to collect multiple errors, which can then be caught in a try-except block and returned as an IssueList.

Source code in cognite/neat/issues/_base.py 
390
391
392
393
394
395
396
397
398
399
class MultiValueError(ValueError):
    """This is a container for multiple errors.

    It is used in the pydantic field_validator/model_validator to collect multiple errors, which
    can then be caught in a try-except block and returned as an IssueList.

    """

    def __init__(self, errors: Sequence[NeatIssue]):
        self.errors = list(errors)

NeatError dataclass #

Bases: NeatIssue, Exception

This is the base class for all exceptions (errors) used in Neat.

Source code in cognite/neat/issues/_base.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@dataclass(frozen=True)
class NeatError(NeatIssue, Exception):
    """This is the base class for all exceptions (errors) used in Neat."""

    @classmethod
    def from_pydantic_errors(cls, errors: list[ErrorDetails], **kwargs) -> "list[NeatError]":
        """Convert a list of pydantic errors to a list of Error instances.

        This is intended to be overridden in subclasses to handle specific error types.
        """
        all_errors: list[NeatError] = []
        read_info_by_sheet = kwargs.get("read_info_by_sheet")

        for error in errors:
            ctx = error.get("ctx")
            if isinstance(ctx, dict) and isinstance(multi_error := ctx.get("error"), MultiValueError):
                if read_info_by_sheet:
                    for caught_error in multi_error.errors:
                        cls._adjust_row_numbers(caught_error, read_info_by_sheet)  # type: ignore[arg-type]
                all_errors.extend(multi_error.errors)  # type: ignore[arg-type]
            elif isinstance(ctx, dict) and isinstance(single_error := ctx.get("error"), NeatError):
                if read_info_by_sheet:
                    cls._adjust_row_numbers(single_error, read_info_by_sheet)
                all_errors.append(single_error)
            elif len(error["loc"]) >= 4 and read_info_by_sheet:
                all_errors.append(RowError.from_pydantic_error(error, read_info_by_sheet))
            else:
                all_errors.append(DefaultPydanticError.from_pydantic_error(error))
        return all_errors

    @staticmethod
    def _adjust_row_numbers(caught_error: "NeatError", read_info_by_sheet: dict[str, SpreadsheetRead]) -> None:
        from cognite.neat.issues.errors._properties import PropertyDefinitionDuplicatedError
        from cognite.neat.issues.errors._resources import ResourceNotDefinedError

        reader = read_info_by_sheet.get("Properties", SpreadsheetRead())

        if isinstance(caught_error, PropertyDefinitionDuplicatedError) and caught_error.location_name == "rows":
            adjusted_row_number = (
                tuple(
                    reader.adjusted_row_number(row_no) if isinstance(row_no, int) else row_no
                    for row_no in caught_error.locations or []
                )
                or None
            )
            # The error is frozen, so we have to use __setattr__ to change the row number
            object.__setattr__(caught_error, "locations", adjusted_row_number)
        elif isinstance(caught_error, RowError):
            # Adjusting the row number to the actual row number in the spreadsheet
            new_row = reader.adjusted_row_number(caught_error.row)
            # The error is frozen, so we have to use __setattr__ to change the row number
            object.__setattr__(caught_error, "row", new_row)
        elif isinstance(caught_error, ResourceNotDefinedError):
            if isinstance(caught_error.row_number, int) and caught_error.sheet_name == "Properties":
                new_row = reader.adjusted_row_number(caught_error.row_number)
                object.__setattr__(caught_error, "row_number", new_row)

from_pydantic_errors(errors, **kwargs) classmethod #

Convert a list of pydantic errors to a list of Error instances.

This is intended to be overridden in subclasses to handle specific error types.

Source code in cognite/neat/issues/_base.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@classmethod
def from_pydantic_errors(cls, errors: list[ErrorDetails], **kwargs) -> "list[NeatError]":
    """Convert a list of pydantic errors to a list of Error instances.

    This is intended to be overridden in subclasses to handle specific error types.
    """
    all_errors: list[NeatError] = []
    read_info_by_sheet = kwargs.get("read_info_by_sheet")

    for error in errors:
        ctx = error.get("ctx")
        if isinstance(ctx, dict) and isinstance(multi_error := ctx.get("error"), MultiValueError):
            if read_info_by_sheet:
                for caught_error in multi_error.errors:
                    cls._adjust_row_numbers(caught_error, read_info_by_sheet)  # type: ignore[arg-type]
            all_errors.extend(multi_error.errors)  # type: ignore[arg-type]
        elif isinstance(ctx, dict) and isinstance(single_error := ctx.get("error"), NeatError):
            if read_info_by_sheet:
                cls._adjust_row_numbers(single_error, read_info_by_sheet)
            all_errors.append(single_error)
        elif len(error["loc"]) >= 4 and read_info_by_sheet:
            all_errors.append(RowError.from_pydantic_error(error, read_info_by_sheet))
        else:
            all_errors.append(DefaultPydanticError.from_pydantic_error(error))
    return all_errors

NeatIssue dataclass #

This is the base class for all exceptions and warnings (issues) used in Neat.

Source code in cognite/neat/issues/_base.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@total_ordering
@dataclass(frozen=True)
class NeatIssue:
    """This is the base class for all exceptions and warnings (issues) used in Neat."""

    extra: ClassVar[str | None] = None
    fix: ClassVar[str | None] = None

    def as_message(self) -> str:
        """Return a human-readable message for the issue."""
        template = self.__doc__
        if not template:
            return "Missing"
        variables, has_all_optional = self._get_variables()

        msg = template.format(**variables)
        if self.extra and has_all_optional:
            msg += "\n" + self.extra.format(**variables)
        if self.fix:
            msg += f"\nFix: {self.fix.format(**variables)}"
        name = type(self).__name__
        return f"{name}: {msg}"

    def _get_variables(self) -> tuple[dict[str, str], bool]:
        variables: dict[str, str] = {}
        has_all_optional = True
        for name, var_ in vars(self).items():
            if var_ is None:
                has_all_optional = False
            elif isinstance(var_, str):
                variables[name] = var_
            elif isinstance(var_, Path):
                variables[name] = var_.as_posix()
            elif isinstance(var_, Collection):
                variables[name] = humanize_collection(var_)
            else:
                variables[name] = repr(var_)
        return variables, has_all_optional

    def dump(self) -> dict[str, Any]:
        """Return a dictionary representation of the issue."""
        variables = vars(self)
        output = {to_camel(key): self._dump_value(value) for key, value in variables.items() if value is not None}
        output["NeatIssue"] = type(self).__name__
        return output

    @classmethod
    def _dump_value(cls, value: Any) -> list | int | bool | float | str | dict:
        if isinstance(value, str | int | bool | float):
            return value
        elif isinstance(value, frozenset):
            return [cls._dump_value(item) for item in value]
        elif isinstance(value, Path):
            return value.as_posix()
        elif isinstance(value, tuple):
            return [cls._dump_value(item) for item in value]
        elif isinstance(value, ViewId | ContainerId):
            return value.dump(camel_case=True, include_type=True)
        raise ValueError(f"Unsupported type: {type(value)}")

    @classmethod
    def load(cls, data: dict[str, Any]) -> "NeatIssue":
        """Create an instance of the issue from a dictionary."""
        from cognite.neat.issues.errors import _NEAT_ERRORS_BY_NAME, NeatValueError
        from cognite.neat.issues.warnings import _NEAT_WARNINGS_BY_NAME

        if "NeatIssue" not in data:
            raise NeatValueError("The data does not contain a NeatIssue key.")
        issue_type = data.pop("NeatIssue")
        args = {to_snake(key): value for key, value in data.items()}
        if issue_type in _NEAT_ERRORS_BY_NAME:
            return cls._load_values(_NEAT_ERRORS_BY_NAME[issue_type], args)
        elif issue_type in _NEAT_WARNINGS_BY_NAME:
            return cls._load_values(_NEAT_WARNINGS_BY_NAME[issue_type], args)
        else:
            raise NeatValueError(f"Unknown issue type: {issue_type}")

    @classmethod
    def _load_values(cls, neat_issue_cls: "type[NeatIssue]", data: dict[str, Any]) -> "NeatIssue":
        args: dict[str, Any] = {}
        for f in fields(neat_issue_cls):
            if f.name not in data:
                continue
            value = data[f.name]
            args[f.name] = cls._load_value(f.type, value)
        return neat_issue_cls(**args)

    @classmethod
    def _load_value(cls, type_: type, value: Any) -> Any:
        if isinstance(type_, UnionType) or get_origin(type_) is UnionType:
            args = get_args(type_)
            return cls._load_value(args[0], value)
        elif type_ is frozenset or get_origin(type_) is frozenset:
            subtype = get_args(type_)[0]
            return frozenset(cls._load_value(subtype, item) for item in value)
        elif type_ is Path:
            return Path(value)
        elif type_ is tuple or get_origin(type_) is tuple:
            subtype = get_args(type_)[0]
            return tuple(cls._load_value(subtype, item) for item in value)
        elif type_ is ViewId:
            return ViewId.load(value)
        elif type_ is ContainerId:
            return ContainerId.load(value)
        return value

    def __lt__(self, other: "NeatIssue") -> bool:
        if not isinstance(other, NeatIssue):
            return NotImplemented
        return (type(self).__name__, self.as_message()) < (type(other).__name__, other.as_message())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, NeatIssue):
            return NotImplemented
        return (type(self).__name__, self.as_message()) == (type(other).__name__, other.as_message())

as_message() #

Return a human-readable message for the issue.

Source code in cognite/neat/issues/_base.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def as_message(self) -> str:
    """Return a human-readable message for the issue."""
    template = self.__doc__
    if not template:
        return "Missing"
    variables, has_all_optional = self._get_variables()

    msg = template.format(**variables)
    if self.extra and has_all_optional:
        msg += "\n" + self.extra.format(**variables)
    if self.fix:
        msg += f"\nFix: {self.fix.format(**variables)}"
    name = type(self).__name__
    return f"{name}: {msg}"

dump() #

Return a dictionary representation of the issue.

Source code in cognite/neat/issues/_base.py 
101
102
103
104
105
106
def dump(self) -> dict[str, Any]:
    """Return a dictionary representation of the issue."""
    variables = vars(self)
    output = {to_camel(key): self._dump_value(value) for key, value in variables.items() if value is not None}
    output["NeatIssue"] = type(self).__name__
    return output

load(data) classmethod #

Create an instance of the issue from a dictionary.

Source code in cognite/neat/issues/_base.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@classmethod
def load(cls, data: dict[str, Any]) -> "NeatIssue":
    """Create an instance of the issue from a dictionary."""
    from cognite.neat.issues.errors import _NEAT_ERRORS_BY_NAME, NeatValueError
    from cognite.neat.issues.warnings import _NEAT_WARNINGS_BY_NAME

    if "NeatIssue" not in data:
        raise NeatValueError("The data does not contain a NeatIssue key.")
    issue_type = data.pop("NeatIssue")
    args = {to_snake(key): value for key, value in data.items()}
    if issue_type in _NEAT_ERRORS_BY_NAME:
        return cls._load_values(_NEAT_ERRORS_BY_NAME[issue_type], args)
    elif issue_type in _NEAT_WARNINGS_BY_NAME:
        return cls._load_values(_NEAT_WARNINGS_BY_NAME[issue_type], args)
    else:
        raise NeatValueError(f"Unknown issue type: {issue_type}")

NeatIssueList #

Bases: UserList[T_NeatIssue], Sequence[T_NeatIssue], ABC

This is a generic list of NeatIssues.

Source code in cognite/neat/issues/_base.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
class NeatIssueList(UserList[T_NeatIssue], Sequence[T_NeatIssue], ABC):
    """This is a generic list of NeatIssues."""

    def __init__(self, issues: Sequence[T_NeatIssue] | None = None, title: str | None = None):
        super().__init__(issues or [])
        self.title = title

    @property
    def errors(self) -> Self:
        """Return all the errors in this list."""
        return type(self)([issue for issue in self if isinstance(issue, NeatError)])  # type: ignore[misc]

    @property
    def has_errors(self) -> bool:
        """Return True if this list contains any errors."""
        return any(isinstance(issue, NeatError) for issue in self)

    @property
    def warnings(self) -> Self:
        """Return all the warnings in this list."""
        return type(self)([issue for issue in self if isinstance(issue, NeatWarning)])  # type: ignore[misc]

    def as_errors(self) -> ExceptionGroup:
        """Return an ExceptionGroup with all the errors in this list."""
        return ExceptionGroup(
            "Operation failed",
            [issue for issue in self if isinstance(issue, NeatError)],
        )

    def trigger_warnings(self) -> None:
        """Trigger all warnings in this list."""
        for warning in [issue for issue in self if isinstance(issue, NeatWarning)]:
            warnings.warn(warning, stacklevel=2)

    def to_pandas(self) -> pd.DataFrame:
        """Return a pandas DataFrame representation of this list."""
        return pd.DataFrame([issue.dump() for issue in self])

    def _repr_html_(self) -> str | None:
        return self.to_pandas()._repr_html_()  # type: ignore[operator]

    def as_exception(self) -> "MultiValueError":
        """Return a MultiValueError with all the errors in this list."""
        return MultiValueError(self.errors)

errors: Self property #

Return all the errors in this list.

has_errors: bool property #

Return True if this list contains any errors.

warnings: Self property #

Return all the warnings in this list.

as_errors() #

Return an ExceptionGroup with all the errors in this list.

Source code in cognite/neat/issues/_base.py 
366
367
368
369
370
371
def as_errors(self) -> ExceptionGroup:
    """Return an ExceptionGroup with all the errors in this list."""
    return ExceptionGroup(
        "Operation failed",
        [issue for issue in self if isinstance(issue, NeatError)],
    )

trigger_warnings() #

Trigger all warnings in this list.

Source code in cognite/neat/issues/_base.py 
373
374
375
376
def trigger_warnings(self) -> None:
    """Trigger all warnings in this list."""
    for warning in [issue for issue in self if isinstance(issue, NeatWarning)]:
        warnings.warn(warning, stacklevel=2)

to_pandas() #

Return a pandas DataFrame representation of this list.

Source code in cognite/neat/issues/_base.py 
378
379
380
def to_pandas(self) -> pd.DataFrame:
    """Return a pandas DataFrame representation of this list."""
    return pd.DataFrame([issue.dump() for issue in self])

as_exception() #

Return a MultiValueError with all the errors in this list.

Source code in cognite/neat/issues/_base.py 
385
386
387
def as_exception(self) -> "MultiValueError":
    """Return a MultiValueError with all the errors in this list."""
    return MultiValueError(self.errors)

NeatWarning dataclass #

Bases: NeatIssue, UserWarning

This is the base class for all warnings used in Neat.

Source code in cognite/neat/issues/_base.py 
306
307
308
309
310
311
312
313
@dataclass(frozen=True)
class NeatWarning(NeatIssue, UserWarning):
    """This is the base class for all warnings used in Neat."""

    @classmethod
    def from_warning(cls, warning: WarningMessage) -> "NeatWarning":
        """Create a NeatWarning from a WarningMessage."""
        return DefaultWarning.from_warning_message(warning)

from_warning(warning) classmethod #

Create a NeatWarning from a WarningMessage.

Source code in cognite/neat/issues/_base.py 
310
311
312
313
@classmethod
def from_warning(cls, warning: WarningMessage) -> "NeatWarning":
    """Create a NeatWarning from a WarningMessage."""
    return DefaultWarning.from_warning_message(warning)