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

@@ -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_snake_case
 
 if TYPE_CHECKING:
     from cognite.client import CogniteClient
@@ -285,8 +283,8 @@ def __init__(
 class ViewProperty(CogniteObject, ABC):
     @classmethod
     def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
-        if "direction" in resource:
-            return cast(Self, SingleHopConnectionDefinition.load(resource))
+        if "connectionType" in resource:
+            return cast(Self, ConnectionDefinition.load(resource))
         else:
             return cast(Self, MappedProperty.load(resource))
 
@@ -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,22 @@ def as_apply(self) -> MappedPropertyApply:
 
 
 @dataclass
-class ConnectionDefinition(ViewProperty):
-    ...
+class ConnectionDefinition(ViewProperty, ABC):
+    @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_snake_case(resource["connectionType"])
+
+        if connection_type in {"multi_edge_connection", "single_edge_connection"}:
+            return cast(Self, SingleHopConnectionDefinition.load(resource))
+        elif connection_type in {"single_reverse_direct_relation", "multi_reverse_direct_relation"}:
+            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 +416,7 @@ class SingleHopConnectionDefinition(ConnectionDefinition):
     description: str | None = None
     edge_source: ViewId | None = None
     direction: Literal["outwards", "inwards"] = "outwards"
-    connection_type: Literal["multiEdgeConnection"] = "multiEdgeConnection"
+    connection_type: Literal["single_edge_connection", "multi_edge_connection"] = "multi_edge_connection"
 
     @classmethod
     def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
@@ -446,12 +458,82 @@ def as_apply(self) -> SingleHopConnectionDefinitionApply:
             description=self.description,
             edge_source=self.edge_source,
             direction=self.direction,
+            connection_type=self.connection_type,
         )
 
 
 @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.
+        connection_type (Literal["single_reverse_direct_relation", "multi_reverse_direct_relation"]):
+            The type of connection. The single_reverse_direct_relation type is used to indicate that only a
+            single direct relation is expected to exist. The multi_reverse_direct_relation type is used to
+            indicate that multiple direct relations are expected to exist.
+        name (str | None): Readable property name.
+        description (str | None): Description of the content and suggested use for this property.
+
+    """
+
+    source: ViewId
+    through: ViewPropertyId
+    connection_type: Literal["single_reverse_direct_relation", "multi_reverse_direct_relation"]
+    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"]),
+            connection_type=resource["connectionType"],
+            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,
+            connection_type=self.connection_type,
+        )
+
+
+@dataclass
+class ConnectionDefinitionApply(ViewPropertyApply, ABC):
+    @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_snake_case(resource["connectionType"])
+
+        if connection_type in {"multi_edge_connection", "single_edge_connection"}:
+            return cast(Self, SingleHopConnectionDefinitionApply.load(resource))
+        elif connection_type in {"single_reverse_direct_relation", "multi_reverse_direct_relation"}:
+            return cast(Self, ReverseSingleHopConnectionApply.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
 
 
 T_ConnectionDefinitionApply = TypeVar("T_ConnectionDefinitionApply", bound=ConnectionDefinitionApply)
@@ -465,7 +547,7 @@ class SingleHopConnectionDefinitionApply(ConnectionDefinitionApply):
     description: str | None = None
     edge_source: ViewId | None = None
     direction: Literal["outwards", "inwards"] = "outwards"
-    connection_type: Literal["multiEdgeConnection"] = "multiEdgeConnection"
+    connection_type: Literal["single_edge_connection", "multi_edge_connection"] = "multi_edge_connection"
 
     @classmethod
     def _load(cls, resource: dict[str, Any], cognite_client: CogniteClient | None = None) -> Self:
@@ -500,3 +582,49 @@ 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.
+        connection_type (Literal["single_reverse_direct_relation", "multi_reverse_direct_relation"]):
+            The type of connection. The single_reverse_direct_relation type is used to indicate that only a
+            single direct relation is expected to exist. The multi_reverse_direct_relation type is used to
+            indicate that multiple direct relations are expected to exist.
+        name (str | None): Readable property name.
+        description (str | None): Description of the content and suggested use for this property.
+
+    """
+
+    source: ViewId
+    through: ViewPropertyId
+    connection_type: Literal["single_reverse_direct_relation", "multi_reverse_direct_relation"]
+    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"]),
+            connection_type=resource["connectionType"],
+            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,
+            "connectionType" if camel_case else "connection_type": self.connection_type,
+        }

tests/tests_unit/test_data_classes/test_data_models/test_views.py

@@ -54,7 +54,7 @@ def test_load_dumped_mapped_property_for_apply(self) -> None:
 
     def test_load_dump_connection_property(self) -> None:
         input = {
-            "connectionType": "multiEdgeConnection",
+            "connectionType": "multi_edge_connection",
             "type": {"space": "mySpace", "externalId": "myExternalId"},
             "source": {"type": "view", "space": "mySpace", "externalId": "myExternalId", "version": "myVersion"},
             "direction": "outwards",
@@ -64,7 +64,7 @@ def test_load_dump_connection_property(self) -> None:
         actual = ViewProperty.load(input)
 
         assert actual.dump(camel_case=False) == {
-            "connection_type": "multiEdgeConnection",
+            "connection_type": "multi_edge_connection",
             "description": None,
             "direction": "outwards",
             "edge_source": None,