feat(alpha): record view CRUD support

cognite/client/_api/data_modeling/views.py

@@ -3,22 +3,39 @@
 import asyncio
 from collections import defaultdict
 from collections.abc import AsyncIterator, Sequence
-from typing import TYPE_CHECKING, Literal, cast, overload
+from typing import TYPE_CHECKING, Any, Literal, cast, overload
 
 from cognite.client._api_client import APIClient
 from cognite.client._constants import DATA_MODELING_DEFAULT_LIMIT_READ
+from cognite.client.data_classes._base import CogniteResource
 from cognite.client.data_classes.data_modeling.ids import (
     ViewId,
     ViewIdentifier,
     _load_identifier,
 )
-from cognite.client.data_classes.data_modeling.views import View, ViewApply, ViewFilter, ViewList
+from cognite.client.data_classes.data_modeling.views import (
+    RecordViewApply,
+    View,
+    ViewApply,
+    ViewFilter,
+    ViewList,
+    ViewUsedFor,
+)
+from cognite.client.utils._experimental import FeaturePreviewWarning
 
 if TYPE_CHECKING:
     from cognite.client import AsyncCogniteClient
     from cognite.client.config import ClientConfig
 
 
+class _ViewOrRecordViewApplyAdapter(CogniteResource):
+    @classmethod
+    def _load(cls, resource: dict[str, Any]) -> ViewApply | RecordViewApply:  # type: ignore[override]
+        if "streamId" in resource:
+            return RecordViewApply._load(resource)
+        return ViewApply._load(resource)
+
+
 class ViewsAPI(APIClient):
     _RESOURCE_PATH = "/models/views"
 
@@ -27,6 +44,13 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         self._DELETE_LIMIT = 100
         self._RETRIEVE_LIMIT = 100
         self._CREATE_LIMIT = 100
+        self._warn_on_record_views = FeaturePreviewWarning(
+            api_maturity="alpha", sdk_maturity="alpha", feature_name="Views on Records"
+        )
+
+    def _record_views_alpha_headers(self) -> dict[str, str]:
+        self._warn_on_record_views.warn()
+        return {"cdf-version": f"{self._config.api_subversion}-alpha"}
 
     def _get_semaphore(self, operation: Literal["read_schema", "write_schema"]) -> asyncio.BoundedSemaphore:
         from cognite.client import global_config
@@ -45,6 +69,7 @@ def __call__(
         include_inherited_properties: bool = True,
         all_versions: bool = False,
         include_global: bool = False,
+        used_for: ViewUsedFor | Sequence[ViewUsedFor] | None = None,
     ) -> AsyncIterator[View]: ...
 
     @overload
@@ -56,6 +81,7 @@ def __call__(
         include_inherited_properties: bool = True,
         all_versions: bool = False,
         include_global: bool = False,
+        used_for: ViewUsedFor | Sequence[ViewUsedFor] | None = None,
     ) -> AsyncIterator[ViewList]: ...
 
     async def __call__(
@@ -66,6 +92,7 @@ async def __call__(
         include_inherited_properties: bool = True,
         all_versions: bool = False,
         include_global: bool = False,
+        used_for: ViewUsedFor | Sequence[ViewUsedFor] | None = None,
     ) -> AsyncIterator[View] | AsyncIterator[ViewList]:
         """Iterate over views
 
@@ -78,18 +105,23 @@ async def __call__(
             include_inherited_properties (bool): Whether to include properties inherited from views this view implements.
             all_versions (bool): Whether to return all versions. If false, only the newest version is returned, which is determined based on the 'createdTime' field.
             include_global (bool): Whether to include global views.
+            used_for (ViewUsedFor | Sequence[ViewUsedFor] | None): Only return views used for the given
+                type(s). Passing "record" is an alpha feature, subject to breaking
+                changes without prior notice.
 
         Yields:
             View | ViewList: yields View one by one if chunk_size is not specified, else ViewList objects.
         """  # noqa: DOC404
-        filter_ = ViewFilter(space, include_inherited_properties, all_versions, include_global)
+        filter_ = ViewFilter(space, include_inherited_properties, all_versions, include_global, used_for)
+        headers = self._record_views_alpha_headers() if filter_.used_for and "record" in filter_.used_for else None
         async for item in self._list_generator(
             list_cls=ViewList,
             resource_cls=View,
             method="GET",
             chunk_size=chunk_size,
             limit=limit,
             filter=filter_.dump(camel_case=True),
+            headers=headers,
             semaphore=self._get_semaphore("read_schema"),
         ):
             yield item
@@ -174,6 +206,7 @@ async def list(
         include_inherited_properties: bool = True,
         all_versions: bool = False,
         include_global: bool = False,
+        used_for: ViewUsedFor | Sequence[ViewUsedFor] | None = None,
     ) -> ViewList:
         """`List views <https://api-docs.cognite.com/20230101/tag/Views/operation/listViews>`_.
 
@@ -183,6 +216,9 @@ async def list(
             include_inherited_properties (bool): Whether to include properties inherited from views this view implements.
             all_versions (bool): Whether to return all versions. If false, only the newest version is returned, which is determined based on the 'createdTime' field.
             include_global (bool): Whether to include global views.
+            used_for (ViewUsedFor | Sequence[ViewUsedFor] | None): Only return views used for the given
+                type(s). Passing "record" is an alpha feature, subject to breaking
+                changes without prior notice.
 
         Returns:
             ViewList: List of requested views
@@ -206,28 +242,30 @@ async def list(
                 >>> for view_list in client.data_modeling.views(chunk_size=10):
                 ...     view_list  # do something with the views
         """
-        filter_ = ViewFilter(space, include_inherited_properties, all_versions, include_global)
+        filter_ = ViewFilter(space, include_inherited_properties, all_versions, include_global, used_for)
+        headers = self._record_views_alpha_headers() if filter_.used_for and "record" in filter_.used_for else None
 
         return await self._list(
             list_cls=ViewList,
             resource_cls=View,
             method="GET",
             limit=limit,
             filter=filter_.dump(camel_case=True),
+            headers=headers,
             override_semaphore=self._get_semaphore("read_schema"),
         )
 
     @overload
-    async def apply(self, view: Sequence[ViewApply]) -> ViewList: ...
+    async def apply(self, view: Sequence[ViewApply | RecordViewApply]) -> ViewList: ...
 
     @overload
-    async def apply(self, view: ViewApply) -> View: ...
+    async def apply(self, view: ViewApply | RecordViewApply) -> View: ...
 
-    async def apply(self, view: ViewApply | Sequence[ViewApply]) -> View | ViewList:
+    async def apply(self, view: ViewApply | RecordViewApply | Sequence[ViewApply | RecordViewApply]) -> View | ViewList:
         """`Create or update (upsert) one or more views <https://api-docs.cognite.com/20230101/tag/Views/operation/ApplyViews>`_.
 
         Args:
-            view (ViewApply | Sequence[ViewApply]): View(s) to create or update.
+            view (ViewApply | RecordViewApply | Sequence[ViewApply | RecordViewApply]): View(s) to create or update.
 
         Returns:
             View | ViewList: Created view(s)
@@ -311,11 +349,35 @@ async def apply(self, view: ViewApply | Sequence[ViewApply]) -> View | ViewList:
                 ...     },
                 ... )
                 >>> res = client.data_modeling.views.apply([work_order_view, asset_view])
+
+            Create a record-backed view (alpha feature, subject to breaking changes without prior notice):
+
+                >>> from cognite.client.data_classes.data_modeling import (
+                ...     ContainerId,
+                ...     MappedPropertyApply,
+                ...     RecordViewApply,
+                ... )
+                >>> record_view = RecordViewApply(
+                ...     space="mySpace",
+                ...     external_id="myRecordView",
+                ...     version="v1",
+                ...     stream_id="myStream",
+                ...     properties={
+                ...         "title": MappedPropertyApply(
+                ...             container=ContainerId("mySpace", "myRecordContainer"),
+                ...             container_property_identifier="title",
+                ...         ),
+                ...     },
+                ... )
+                >>> res = client.data_modeling.views.apply(record_view)
         """
+        views = [view] if isinstance(view, (ViewApply, RecordViewApply)) else view
+        headers = self._record_views_alpha_headers() if any(isinstance(v, RecordViewApply) for v in views) else None
         return await self._create_multiple(
             list_cls=ViewList,
             resource_cls=View,
             items=view,
-            input_resource_cls=ViewApply,
+            input_resource_cls=_ViewOrRecordViewApplyAdapter,
+            headers=headers,
             override_semaphore=self._get_semaphore("write_schema"),
         )

cognite/client/data_classes/data_modeling/__init__.py

@@ -157,6 +157,7 @@
     MultiEdgeConnectionApply,
     MultiReverseDirectRelation,
     MultiReverseDirectRelationApply,
+    RecordViewApply,
     SingleHopConnectionDefinition,
     View,
     ViewApply,
@@ -265,6 +266,7 @@
     "RecordSourceSelector",
     "RecordTargetUnit",
     "RecordTargetUnits",
+    "RecordViewApply",
     "RecordWrite",
     "RecordWriteList",
     "RequiresConstraint",