Errors

cognite.neat._issues.errors

DefaultPydanticError dataclass

Bases: NeatError, ValueError

{type}: {msg} [loc={loc}]

Source code in cognite/neat/_issues/_base.py

@dataclass(unsafe_hash=True)
class DefaultPydanticError(NeatError, ValueError):
    """{type}: {msg} [loc={loc}]"""

    type: str
    loc: tuple[int | str, ...]
    msg: str

    @classmethod
    def from_pydantic_error(cls, error: ErrorDetails) -> "DefaultPydanticError":
        return cls(
            type=error["type"],
            loc=error["loc"],
            msg=error["msg"],
        )

    def as_message(self, include_type: bool = True) -> str:
        if self.loc and len(self.loc) == 1:
            return f"{self.loc[0]} sheet: {self.msg}"
        elif self.loc and len(self.loc) == 2:
            return f"{self.loc[0]} sheet field/column <{self.loc[1]}>: {self.msg}"
        else:
            return self.msg

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

@dataclass(unsafe_hash=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:
            if error["type"] == "is_instance_of" and error["loc"][1] == "is-instance[SheetList]":
                # Skip the error for SheetList, as it is not relevant for the user. This is an
                # internal class used to have helper methods for a lists as .to_pandas()
                continue
            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

@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:
        if error["type"] == "is_instance_of" and error["loc"][1] == "is-instance[SheetList]":
            # Skip the error for SheetList, as it is not relevant for the user. This is an
            # internal class used to have helper methods for a lists as .to_pandas()
            continue
        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

RowError dataclass

Bases: NeatError, ValueError

In {sheet_name}, row={row}, column={column}: {msg}. [type={type}, input_value={input}]

Source code in cognite/neat/_issues/_base.py

@dataclass(unsafe_hash=True)
class RowError(NeatError, ValueError):
    """In {sheet_name}, row={row}, column={column}: {msg}. [type={type}, input_value={input}]"""

    extra = "For further information visit {url}"

    sheet_name: str
    column: str
    row: int
    type: str
    msg: str
    input: Any
    url: str | None = None

    @classmethod
    def from_pydantic_error(
        cls,
        error: ErrorDetails,
        read_info_by_sheet: dict[str, SpreadsheetRead] | None = None,
    ) -> Self:
        sheet_name, _, row, column, *__ = error["loc"]
        reader = (read_info_by_sheet or {}).get(str(sheet_name), SpreadsheetRead())
        return cls(
            sheet_name=str(sheet_name),
            column=str(column),
            row=reader.adjusted_row_number(int(row)),
            type=error["type"],
            msg=error["msg"],
            input=error.get("input"),
            url=str(url) if (url := error.get("url")) else None,
        )

    def as_message(self, include_type: bool = True) -> str:
        input_str = str(self.input) if self.input is not None else ""
        input_str = input_str[:50] + "..." if len(input_str) > 50 else input_str
        output = (
            f"In {self.sheet_name}, row={self.row}, column={self.column}: {self.msg}. "
            f"[type={self.type}, input_value={input_str}]"
        )
        if self.url:
            output += f" For further information visit {self.url}"
        return output

AuthorizationError dataclass

Bases: NeatError, RuntimeError

Missing authorization for {action}: {reason}

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class AuthorizationError(NeatError, RuntimeError):
    """Missing authorization for {action}: {reason}"""

    action: str
    reason: str

FileMissingRequiredFieldError dataclass

Bases: NeatError, ValueError

Missing required {field_name} in {filepath}: {field}

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class FileMissingRequiredFieldError(NeatError, ValueError):
    """Missing required {field_name} in {filepath}: {field}"""

    filepath: Path
    field_name: str
    field: str

FileNotAFileError dataclass

Bases: NeatError, FileNotFoundError

{filepath} is not a file

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class FileNotAFileError(NeatError, FileNotFoundError):
    """{filepath} is not a file"""

    fix = "Make sure to provide a valid file"
    filepath: Path

FileNotFoundNeatError dataclass

Bases: NeatError, FileNotFoundError

File {filepath} not found

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class FileNotFoundNeatError(NeatError, FileNotFoundError):
    """File {filepath} not found"""

    fix = "Make sure to provide a valid file"
    filepath: Path

FileReadError dataclass

Bases: NeatError, RuntimeError

Error when reading file, {filepath}: {reason}

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class FileReadError(NeatError, RuntimeError):
    """Error when reading file, {filepath}: {reason}"""

    fix = "Is the {filepath} open in another program? Is the file corrupted?"
    filepath: Path
    reason: str

FileTypeUnexpectedError dataclass

Bases: NeatError, TypeError

Unexpected file type: {filepath}. Expected format: {expected_format}

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class FileTypeUnexpectedError(NeatError, TypeError):
    """Unexpected file type: {filepath}. Expected format: {expected_format}"""

    filepath: Path
    expected_format: frozenset[str]

NeatYamlError dataclass

Bases: NeatError, YAMLError

Invalid YAML: {reason}

Source code in cognite/neat/_issues/errors/_external.py

@dataclass(unsafe_hash=True)
class NeatYamlError(NeatError, YAMLError):
    """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

NeatImportError dataclass

Bases: NeatError, ImportError

The functionality requires {module}. You can include it in your neat installation with pip install "cognite-neat[{neat_extra}]".

Source code in cognite/neat/_issues/errors/_general.py

@dataclass(unsafe_hash=True)
class NeatImportError(NeatError, ImportError):
    """The functionality requires {module}. You can include it
    in your neat installation with `pip install "cognite-neat[{neat_extra}]"`.
    """

    module: str
    neat_extra: str

NeatTypeError dataclass

Bases: NeatError, TypeError

{raw_message}

Source code in cognite/neat/_issues/errors/_general.py

@dataclass(unsafe_hash=True)
class NeatTypeError(NeatError, TypeError):
    """{raw_message}"""

    raw_message: str

NeatValueError dataclass

Bases: NeatError, ValueError

{raw_message}

Source code in cognite/neat/_issues/errors/_general.py

@dataclass(unsafe_hash=True)
class NeatValueError(NeatError, ValueError):
    """{raw_message}"""

    raw_message: str

RegexViolationError dataclass

Bases: NeatError, ValueError

Value, {value} in {location} failed regex, {regex}, validation. Make sure that the name follows the regex pattern.

Source code in cognite/neat/_issues/errors/_general.py

@dataclass(unsafe_hash=True)
class RegexViolationError(NeatError, ValueError):
    """Value, {value} in {location} failed regex, {regex}, validation.
    Make sure that the name follows the regex pattern."""

    value: str
    regex: str
    location: str

PropertyDefinitionDuplicatedError dataclass

Bases: PropertyError[T_Identifier]

The {resource_type} with identifier {identifier} has multiple definitions for the property {property_name} with values {property_values}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class PropertyDefinitionDuplicatedError(PropertyError[T_Identifier]):
    """The {resource_type} with identifier {identifier} has multiple definitions for the property {property_name}
    with values {property_values}
    """

    extra = "in locations {locations} with name {location_name}"

    property_values: frozenset[str | int | float | bool | None | tuple[str | int | float | bool | None, ...]]
    locations: tuple[str | int, ...] | None = None
    location_name: str | None = None

PropertyDefinitionError dataclass

Bases: PropertyError[T_Identifier]

Invalid property definition for {resource_type} {identifier}.{property_name}: {reason}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class PropertyDefinitionError(PropertyError[T_Identifier]):
    """Invalid property definition for {resource_type} {identifier}.{property_name}: {reason}"""

    reason: str

PropertyMappingDuplicatedError dataclass

Bases: PropertyError[T_Identifier], Generic[T_Identifier, T_ReferenceIdentifier]

The {resource_type} with identifier {identifier}.{property_name} is mapped to by: {mappings}. Ensure that only one {mapping_type} maps to {resource_type} {identifier}.{property_name}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class PropertyMappingDuplicatedError(PropertyError[T_Identifier], Generic[T_Identifier, T_ReferenceIdentifier]):
    """The {resource_type} with identifier {identifier}.{property_name} is mapped to by: {mappings}. Ensure
    that only one {mapping_type} maps to {resource_type} {identifier}.{property_name}"""

    mappings: frozenset[T_ReferenceIdentifier]
    mapping_type: ResourceType

PropertyNotFoundError dataclass

Bases: PropertyError, Generic[T_Identifier, T_ReferenceIdentifier]

The {resource_type} with identifier {identifier} does not have a property {property_name}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class PropertyNotFoundError(PropertyError, Generic[T_Identifier, T_ReferenceIdentifier]):
    """The {resource_type} with identifier {identifier} does not have a property {property_name}"""

    extra = "referred to by {referred_type} {referred_by} does not exist"
    fix = "Ensure the {resource_type} {identifier} has a {property_name} property"

    referred_by: T_ReferenceIdentifier | None = None
    referred_type: ResourceType | None = None

PropertyTypeNotSupportedError dataclass

Bases: PropertyError[T_Identifier]

The {resource_type} with identifier {identifier} has a property {property_name} of unsupported type {property_type}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class PropertyTypeNotSupportedError(PropertyError[T_Identifier]):
    """The {resource_type} with identifier {identifier} has a property {property_name}
    of unsupported type {property_type}"""

    property_type: str

ReversedConnectionNotFeasibleError dataclass

Bases: PropertyError[T_Identifier]

The {resource_type} {property_name} with identifier {identifier} of the view {target_view_id} cannot be made since view {source_view_id} does not have direct connection {direct_connection} defined, or {direct_connection} value type is not {target_view_id}

Source code in cognite/neat/_issues/errors/_properties.py

@dataclass(unsafe_hash=True)
class ReversedConnectionNotFeasibleError(PropertyError[T_Identifier]):
    """The {resource_type} {property_name} with identifier {identifier} of the view {target_view_id} cannot be made
    since view {source_view_id} does not have direct connection {direct_connection} defined,
    or {direct_connection} value type is not {target_view_id}
    """

    target_view_id: str
    source_view_id: str
    direct_connection: str

ResourceChangedError dataclass

Bases: ResourceError[T_Identifier]

The {resource_type} with identifier {identifier} has changed{changed}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceChangedError(ResourceError[T_Identifier]):
    """The {resource_type} with identifier {identifier} has changed{changed}"""

    fix = (
        "When extending model with extension set to addition or reshape, "
        "the {resource_type} properties must remain the same"
    )

    changed_properties: frozenset[str]
    changed_attributes: frozenset[str]

    def as_message(self, include_type: bool = True) -> str:
        if self.changed_properties:
            changed = f" properties {humanize_collection(self.changed_properties)}."
        elif self.changed_attributes:
            changed = f" attributes {humanize_collection(self.changed_attributes)}."
        else:
            changed = "."
        msg = (self.__doc__ or "").format(resource_type=self.resource_type, identifier=self.identifier, changed=changed)
        msg += f"Fix {self.fix.format(resource_type=self.resource_type)}"
        return msg

ResourceConvertionError dataclass

Bases: ResourceError, ValueError

Failed to convert the {resource_type} {identifier} to {target_format}: {reason}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceConvertionError(ResourceError, ValueError):
    """Failed to convert the {resource_type} {identifier} to {target_format}: {reason}"""

    fix = "Check the error message and correct the rules."
    target_format: str
    reason: str

ResourceCreationError dataclass

Bases: ResourceError[T_Identifier], ValueError

Failed to create {resource_type} with identifier {identifier}. The error was: {error}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceCreationError(ResourceError[T_Identifier], ValueError):
    """Failed to create {resource_type} with identifier {identifier}. The error was: {error}"""

    error: str

ResourceDuplicatedError dataclass

Bases: ResourceError[T_Identifier]

The {resource_type} with identifier {identifier} is duplicated in {location}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceDuplicatedError(ResourceError[T_Identifier]):
    """The {resource_type} with identifier {identifier} is duplicated in {location}"""

    fix = "Remove the duplicate {resource_type} {identifier}."
    location: str

ResourceError dataclass

Bases: NeatError, Generic[T_Identifier], RuntimeError

Base class for resource errors {resource_type} with identifier {identifier}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceError(NeatError, Generic[T_Identifier], RuntimeError):
    """Base class for resource errors {resource_type} with identifier {identifier}"""

    identifier: T_Identifier
    resource_type: ResourceType

ResourceMissingIdentifierError dataclass

Bases: NeatError, ValueError

The {resource_type} with name {name} is missing an identifier.

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceMissingIdentifierError(NeatError, ValueError):
    """The {resource_type} with name {name} is missing an identifier."""

    resource_type: ResourceType
    name: str | None = None

ResourceNotDefinedError dataclass

Bases: ResourceError[T_Identifier]

The {resource_type} {identifier} is not defined in the {location}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceNotDefinedError(ResourceError[T_Identifier]):
    """The {resource_type} {identifier} is not defined in the {location}"""

    extra = "{column_name} {row_number} in {sheet_name}"
    fix = "Define the {resource_type} {identifier} in {location}."

    location: str
    column_name: str | None = None
    row_number: int | None = None
    sheet_name: str | None = None

ResourceNotFoundError dataclass

Bases: ResourceError, Generic[T_Identifier, T_ReferenceIdentifier]

The {resource_type} with identifier {identifier} does not exist

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceNotFoundError(ResourceError, Generic[T_Identifier, T_ReferenceIdentifier]):
    """The {resource_type} with identifier {identifier} does not exist"""

    extra = " This is expected by {referred_type} {referred_by}."

    fix = "Create the {resource_type} {identifier}"

    referred_by: T_ReferenceIdentifier | None = None
    referred_type: ResourceType | None = None
    more: str | None = None

    def as_message(self, include_type: bool = True) -> str:
        msg = (self.__doc__ or "").format(resource_type=self.resource_type, identifier=self.identifier)
        if self.referred_by and self.referred_type:
            msg += self.extra.format(referred_type=self.referred_type, referred_by=self.referred_by)
        if self.more:
            msg += f" {self.more}"
        msg += f" Fix {self.fix.format(resource_type=self.resource_type, identifier=self.identifier)}"
        return msg

ResourceRetrievalError dataclass

Bases: ResourceError[T_Identifier]

Failed to retrieve {resource_type} with identifier {identifier}. The error was: {error}

Source code in cognite/neat/_issues/errors/_resources.py

@dataclass(unsafe_hash=True)
class ResourceRetrievalError(ResourceError[T_Identifier]):
    """Failed to retrieve {resource_type} with identifier {identifier}. The error was: {error}"""

    error: str

WorkflowConfigurationNotSetError dataclass

Bases: NeatError, RuntimeError

The configuration variable '{config_variable}' is not set. Please set the configuration before running the workflow.

Source code in cognite/neat/_issues/errors/_workflow.py

@dataclass(unsafe_hash=True)
class WorkflowConfigurationNotSetError(NeatError, RuntimeError):
    """The configuration variable '{config_variable}' is not set. Please set the configuration
    before running the workflow."""

    config_variable: str

WorkFlowMissingDataError dataclass

Bases: NeatError, ValueError

In the workflow step {step_name} the following data is missing: {missing_data}.

Source code in cognite/neat/_issues/errors/_workflow.py

@dataclass(unsafe_hash=True)
class WorkFlowMissingDataError(NeatError, ValueError):
    """In the workflow step {step_name} the following data is missing: {missing_data}."""

    step_name: str
    missing_data: frozenset[str]

WorkflowStepNotInitializedError dataclass

Bases: NeatError, RuntimeError

Step {step_name} has not been initialized.

Source code in cognite/neat/_issues/errors/_workflow.py

@dataclass(unsafe_hash=True)
class WorkflowStepNotInitializedError(NeatError, RuntimeError):
    """Step {step_name} has not been initialized."""

    step_name: str

WorkflowStepOutputError dataclass

Bases: NeatError, RuntimeError

Object type {step_type} is not supported as step output.

Step output must be of type DataContract or a FlowMessage.

Source code in cognite/neat/_issues/errors/_workflow.py

@dataclass(unsafe_hash=True)
class WorkflowStepOutputError(NeatError, RuntimeError):
    """Object type {step_type} is not supported as step output.

    Step output must be of type DataContract or a FlowMessage.
    """

    step_type: str