diff --git a/cognite/client/data_classes/data_modeling/ids.py b/cognite/client/data_classes/data_modeling/ids.py index 385b5d46b0..ce2eb2efe7 100644 --- a/cognite/client/data_classes/data_modeling/ids.py +++ b/cognite/client/data_classes/data_modeling/ids.py @@ -2,13 +2,19 @@ from abc import ABC from dataclasses import asdict, dataclass, field -from typing import Any, ClassVar, Literal, Protocol, Sequence, Tuple, TypeVar, Union, cast +from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, Sequence, Tuple, TypeVar, Union, cast +from typing_extensions import Self + +from cognite.client.data_classes._base import CogniteObject from cognite.client.utils._auxiliary import rename_and_exclude_keys from cognite.client.utils._identifier import DataModelingIdentifier, DataModelingIdentifierSequence from cognite.client.utils._text import convert_all_keys_recursive, convert_all_keys_to_snake_case from cognite.client.utils.useful_types import SequenceNotStr +if TYPE_CHECKING: + from cognite.client import CogniteClient + @dataclass(frozen=True) class AbstractDataclass(ABC): @@ -138,6 +144,25 @@ def as_property_ref(self, property: str) -> tuple[str, ...]: return (self.space, self.as_source_identifier(), property) +@dataclass(frozen=True) +class ViewPropertyId(CogniteObject): + view_id: ViewId + property: str + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + view_id=ViewId.load(resource["source"]), + property=resource["identifier"], + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return { + "source": self.view_id.dump(camel_case=camel_case, include_type=True), + "identifier": self.property, + } + + @dataclass(frozen=True) class DataModelId(VersionedDataModelingId): _type = "datamodel" diff --git a/cognite/client/data_classes/data_modeling/views.py b/cognite/client/data_classes/data_modeling/views.py index 4ecab2c20d..e15e1b5c67 100644 --- a/cognite/client/data_classes/data_modeling/views.py +++ b/cognite/client/data_classes/data_modeling/views.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod from dataclasses import asdict, dataclass -from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast +from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeVar, cast from typing_extensions import Self @@ -18,11 +18,9 @@ DirectRelationReference, PropertyType, ) -from cognite.client.data_classes.data_modeling.ids import ContainerId, ViewId +from cognite.client.data_classes.data_modeling.ids import ContainerId, ViewId, ViewPropertyId from cognite.client.data_classes.filters import Filter -from cognite.client.utils._text import ( - convert_all_keys_to_camel_case_recursive, -) +from cognite.client.utils._text import convert_all_keys_to_camel_case_recursive, to_camel_case if TYPE_CHECKING: from cognite.client import CogniteClient @@ -298,8 +296,8 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]: class ViewPropertyApply(CogniteObject, ABC): @classmethod def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: - if "direction" in resource: - return cast(Self, SingleHopConnectionDefinitionApply.load(resource)) + if "connectionType" in resource: + return cast(Self, ConnectionDefinitionApply.load(resource)) else: return cast(Self, MappedPropertyApply.load(resource)) @@ -392,8 +390,24 @@ def as_apply(self) -> MappedPropertyApply: @dataclass -class ConnectionDefinition(ViewProperty): - ... +class ConnectionDefinition(ViewProperty, ABC): + connection_type: ClassVar[Literal["multiEdgeConnection", "singleReverseDirectionRelation"]] + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + if "connectionType" not in resource: + raise ValueError(f"{cls.__name__} must have a connectionType") + connection_type = to_camel_case(resource["connectionType"]) + + if connection_type == "multiEdgeConnection": + return cast(Self, SingleHopConnectionDefinition.load(resource)) + elif connection_type == "singleReverseDirectRelation": + return cast(Self, ReverseSingleHopConnection.load(resource)) + else: + raise ValueError(f"Cannot load {cls.__name__}: Unknown connection type {connection_type}") + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + raise NotImplementedError @dataclass @@ -404,7 +418,6 @@ class SingleHopConnectionDefinition(ConnectionDefinition): description: str | None = None edge_source: ViewId | None = None direction: Literal["outwards", "inwards"] = "outwards" - connection_type: Literal["multiEdgeConnection"] = "multiEdgeConnection" @classmethod def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: @@ -417,8 +430,6 @@ def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = ) if "direction" in resource: instance.direction = resource["direction"] - if "connectionType" in resource: - instance.connection_type = resource["connectionType"] return instance def dump(self, camel_case: bool = True) -> dict[str, Any]: @@ -450,8 +461,56 @@ def as_apply(self) -> SingleHopConnectionDefinitionApply: @dataclass -class ConnectionDefinitionApply(ViewPropertyApply): - ... +class ReverseSingleHopConnection(ConnectionDefinition): + """Describes the direct relation(s) pointing to instances read through this view. This connection type is used to + aid in discovery and documentation of the view + + It is called 'ReverseDirectRelationConnection' in the API spec. + + Args: + source (ViewId): The node(s) containing the direct relation property can be read through + the view specified in 'source'. + through (ViewPropertyId): The view or container of the node containing the direct relation property. + name (str | None): Readable property name. + description (str | None): Description of the content and suggested use for this property. + + """ + + source: ViewId + through: ViewPropertyId + name: str | None = None + description: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + source=ViewId.load(resource["source"]), + through=ViewPropertyId.load(resource["through"]), + name=resource.get("name"), + description=resource.get("description"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return { + "source": self.source.dump(camel_case), + "through": self.through.dump(camel_case), + "name": self.name, + "description": self.description, + "connectionType" if camel_case else "connection_type": self.connection_type, + } + + def as_apply(self) -> ReverseSingleHopConnectionApply: + return ReverseSingleHopConnectionApply( + source=self.source, + through=self.through, + name=self.name, + description=self.description, + ) + + +@dataclass +class ConnectionDefinitionApply(ViewPropertyApply, ABC): + connection_type: ClassVar[Literal["multiEdgeConnection", "singleReverseDirectRelation"]] T_ConnectionDefinitionApply = TypeVar("T_ConnectionDefinitionApply", bound=ConnectionDefinitionApply) @@ -465,7 +524,6 @@ class SingleHopConnectionDefinitionApply(ConnectionDefinitionApply): description: str | None = None edge_source: ViewId | None = None direction: Literal["outwards", "inwards"] = "outwards" - connection_type: Literal["multiEdgeConnection"] = "multiEdgeConnection" @classmethod def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: @@ -478,8 +536,6 @@ def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = ) if "direction" in resource: instance.direction = resource["direction"] - if "connectionType" in resource: - instance.connection_type = resource["connectionType"] return instance def dump(self, camel_case: bool = True) -> dict: @@ -500,3 +556,42 @@ def dump(self, camel_case: bool = True) -> dict: output[("connectionType" if camel_case else "connection_type")] = self.connection_type return output + + +@dataclass +class ReverseSingleHopConnectionApply(ConnectionDefinitionApply): + """Describes the direct relation(s) pointing to instances read through this view. This connection type is used to + aid in discovery and documentation of the view + + It is called 'ReverseDirectRelationConnection' in the API spec. + + Args: + source (ViewId): The node(s) containing the direct relation property can be read through + the view specified in 'source'. + through (ViewPropertyId): The view or container of the node containing the direct relation property. + name (str | None): Readable property name. + description (str | None): Description of the content and suggested use for this property. + + """ + + source: ViewId + through: ViewPropertyId + name: str | None = None + description: str | None = None + + @classmethod + def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self: + return cls( + source=ViewId.load(resource["source"]), + through=ViewPropertyId.load(resource["through"]), + name=resource.get("name"), + description=resource.get("description"), + ) + + def dump(self, camel_case: bool = True) -> dict[str, Any]: + return { + "source": self.source.dump(camel_case, include_type=True), + "through": self.through.dump(camel_case), + "name": self.name, + "description": self.description, + }