feat: Implemented ReverseDirectRelationConnection

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"

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,
+        }