feat(streams): Streams API

cognite/client/_api/data_modeling/__init__.py

@@ -9,6 +9,7 @@
 from cognite.client._api.data_modeling.instances import InstancesAPI
 from cognite.client._api.data_modeling.spaces import SpacesAPI
 from cognite.client._api.data_modeling.statistics import StatisticsAPI
+from cognite.client._api.data_modeling.streams import StreamsAPI
 from cognite.client._api.data_modeling.views import ViewsAPI
 from cognite.client._api_client import APIClient
 
@@ -27,6 +28,7 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         self.instances = InstancesAPI(config, api_version, cognite_client)
         self.graphql = DataModelingGraphQLAPI(config, api_version, cognite_client)
         self.statistics = StatisticsAPI(config, api_version, cognite_client)
+        self.streams = StreamsAPI(config, api_version, cognite_client)
 
     def _get_semaphore(
         self, operation: Literal["read", "write", "delete", "search", "read_schema", "write_schema"]

cognite/client/_api/data_modeling/streams.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, overload
+
+from cognite.client._api_client import APIClient
+from cognite.client.data_classes.data_modeling.streams import (
+    Stream,
+    StreamList,
+    StreamWrite,
+)
+from cognite.client.utils._experimental import FeaturePreviewWarning
+from cognite.client.utils._identifier import Identifier, IdentifierSequence
+from cognite.client.utils.useful_types import SequenceNotStr
+
+if TYPE_CHECKING:
+    from cognite.client import AsyncCogniteClient
+    from cognite.client.config import ClientConfig
+
+
+class StreamsAPI(APIClient):
+    _RESOURCE_PATH = "/streams"
+
+    def __init__(self, config: ClientConfig, api_version: str | None, cognite_client: AsyncCogniteClient) -> None:
+        super().__init__(config, api_version, cognite_client)
+        self._CREATE_LIMIT = 1
+        self._DELETE_LIMIT = 1
+        self._warning = FeaturePreviewWarning(
+            api_maturity="General Availability", sdk_maturity="alpha", feature_name="Streams"
+        )
+
+    @overload
+    async def create(self, items: StreamWrite) -> Stream: ...
+
+    @overload
+    async def create(self, items: Sequence[StreamWrite]) -> StreamList: ...
+
+    async def create(self, items: StreamWrite | Sequence[StreamWrite]) -> Stream | StreamList:
+        """`Create streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/createStream>`_.
+
+        Args:
+            items (StreamWrite | Sequence[StreamWrite]): One or more streams to create.
+
+        Returns:
+            Stream | StreamList: The created stream or streams.
+
+        Examples:
+
+            Create a single stream from a template:
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes.data_modeling.streams import (
+                ...     StreamWrite,
+                ...     StreamTemplate,
+                ...     StreamTemplateWriteSettings,
+                ... )
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.create(
+                ...     StreamWrite(
+                ...         external_id="my-stream",
+                ...         settings=StreamTemplateWriteSettings(
+                ...             template=StreamTemplate(name="ImmutableTestStream"),
+                ...         ),
+                ...     )
+                ... )
+        """
+        self._warning.warn()
+        return await self._create_multiple(
+            items=items,
+            list_cls=StreamList,
+            resource_cls=Stream,
+            input_resource_cls=StreamWrite,
+        )
+
+    async def list(self) -> StreamList:
+        """`List streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/listStreams>`_.
+
+        Note:
+            There is no paging limit parameter: the endpoint returns all streams in the project
+            (projects are expected to have few streams).
+
+        Returns:
+            StreamList: The streams in the project.
+
+        Examples:
+
+            List all streams in the project:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.list()
+        """
+        self._warning.warn()
+        return await self._list(method="GET", list_cls=StreamList, resource_cls=Stream)
+
+    async def retrieve(self, external_id: str, include_statistics: bool | None = None) -> Stream | None:
+        """`Retrieve a stream <https://api-docs.cognite.com/20230101/tag/Streams/operation/getStream>`_.
+
+        Args:
+            external_id (str): External ID of the stream to retrieve.
+            include_statistics (bool | None): When ``True``, usage statistics will be returned together
+                with stream settings. Computing statistics can be expensive.
+
+        Returns:
+            Stream | None: The stream metadata (and optionally statistics), or ``None`` if not found.
+
+        Examples:
+
+            Retrieve a stream by external ID:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.retrieve("my-stream")
+
+            Retrieve a stream with usage statistics:
+
+                >>> res = client.data_modeling.streams.retrieve(
+                ...     "my-stream",
+                ...     include_statistics=True,
+                ... )
+        """
+        self._warning.warn()
+        return await self._retrieve(
+            cls=Stream,
+            identifier=Identifier(external_id),
+            params={"includeStatistics": include_statistics} if include_statistics is not None else None,
+        )
+
+    async def delete(self, external_id: str | SequenceNotStr[str]) -> None:
+        """`Delete streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/deleteStreams>`_.
+
+        Note:
+            Deletion is a soft delete that retains capacity for an extended period;
+            prefer deleting only when necessary.
+
+        Args:
+            external_id (str | SequenceNotStr[str]): External ID or list of external IDs of
+                streams to delete.
+
+        Examples:
+
+            Delete a single stream:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> client.data_modeling.streams.delete("my-stream")
+
+            Delete multiple streams:
+
+                >>> client.data_modeling.streams.delete(["stream-a", "stream-b"])
+        """
+        self._warning.warn()
+        await self._delete_multiple(
+            identifiers=IdentifierSequence.load(external_ids=external_id),
+            wrap_ids=True,
+        )

cognite/client/_cognite_client.py

@@ -52,6 +52,7 @@
     from cognite.client._api.data_modeling.space_statistics import SpaceStatisticsAPI
     from cognite.client._api.data_modeling.spaces import SpacesAPI
     from cognite.client._api.data_modeling.statistics import StatisticsAPI
+    from cognite.client._api.data_modeling.streams import StreamsAPI
     from cognite.client._api.data_modeling.views import ViewsAPI
     from cognite.client._api.datapoints import DatapointsAPI
     from cognite.client._api.datapoints_subscriptions import DatapointsSubscriptionAPI
@@ -425,6 +426,7 @@ def _make_accessors_for_building_docs() -> None:
     AsyncCogniteClient.data_modeling.graphql = DataModelingGraphQLAPI  # type: ignore
     AsyncCogniteClient.data_modeling.statistics = StatisticsAPI  # type: ignore
     AsyncCogniteClient.data_modeling.statistics.spaces = SpaceStatisticsAPI  # type: ignore
+    AsyncCogniteClient.data_modeling.streams = StreamsAPI  # type: ignore
     AsyncCogniteClient.documents = DocumentsAPI  # type: ignore
     AsyncCogniteClient.documents.previews = DocumentPreviewAPI  # type: ignore
     AsyncCogniteClient.workflows = WorkflowAPI  # type: ignore

cognite/client/_sync_api/data_modeling/__init__.py

@@ -1,27 +1,23 @@
 """
 ===============================================================================
-c76b2b9351d2a5eee6a710fa9893bfa4
+584030bc5e2a4b8168f54c101f7f521d
 This file is auto-generated from the Async API modules, - do not edit manually!
 ===============================================================================
 """
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
 from cognite.client import AsyncCogniteClient
 from cognite.client._sync_api.data_modeling.containers import SyncContainersAPI
 from cognite.client._sync_api.data_modeling.data_models import SyncDataModelsAPI
 from cognite.client._sync_api.data_modeling.graphql import SyncDataModelingGraphQLAPI
 from cognite.client._sync_api.data_modeling.instances import SyncInstancesAPI
 from cognite.client._sync_api.data_modeling.spaces import SyncSpacesAPI
 from cognite.client._sync_api.data_modeling.statistics import SyncStatisticsAPI
+from cognite.client._sync_api.data_modeling.streams import SyncStreamsAPI
 from cognite.client._sync_api.data_modeling.views import SyncViewsAPI
 from cognite.client._sync_api_client import SyncAPIClient
 
-if TYPE_CHECKING:
-    from cognite.client import AsyncCogniteClient
-
 
 class SyncDataModelingAPI(SyncAPIClient):
     """Auto-generated, do not modify manually."""
@@ -35,3 +31,4 @@ def __init__(self, async_client: AsyncCogniteClient) -> None:
         self.instances = SyncInstancesAPI(async_client)
         self.graphql = SyncDataModelingGraphQLAPI(async_client)
         self.statistics = SyncStatisticsAPI(async_client)
+        self.streams = SyncStreamsAPI(async_client)

cognite/client/_sync_api/data_modeling/streams.py

@@ -0,0 +1,149 @@
+"""
+===============================================================================
+3bfd805fbceb341bb437635ac632d5ad
+This file is auto-generated from the Async API modules, - do not edit manually!
+===============================================================================
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, overload
+
+from cognite.client import AsyncCogniteClient
+from cognite.client._sync_api_client import SyncAPIClient
+from cognite.client.data_classes.data_modeling.streams import Stream, StreamList, StreamWrite
+from cognite.client.utils._async_helpers import run_sync
+from cognite.client.utils.useful_types import SequenceNotStr
+
+if TYPE_CHECKING:
+    from cognite.client import AsyncCogniteClient
+
+
+class SyncStreamsAPI(SyncAPIClient):
+    """Auto-generated, do not modify manually."""
+
+    def __init__(self, async_client: AsyncCogniteClient) -> None:
+        self.__async_client = async_client
+
+    @overload
+    def create(self, items: StreamWrite) -> Stream: ...
+
+    @overload
+    def create(self, items: Sequence[StreamWrite]) -> StreamList: ...
+
+    def create(self, items: StreamWrite | Sequence[StreamWrite]) -> Stream | StreamList:
+        """
+        `Create streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/createStream>`_.
+
+        Args:
+            items (StreamWrite | Sequence[StreamWrite]): One or more streams to create.
+
+        Returns:
+            Stream | StreamList: The created stream or streams.
+
+        Examples:
+
+            Create a single stream from a template:
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes.data_modeling.streams import (
+                ...     StreamWrite,
+                ...     StreamTemplate,
+                ...     StreamTemplateWriteSettings,
+                ... )
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.create(
+                ...     StreamWrite(
+                ...         external_id="my-stream",
+                ...         settings=StreamTemplateWriteSettings(
+                ...             template=StreamTemplate(name="ImmutableTestStream"),
+                ...         ),
+                ...     )
+                ... )
+        """
+        return run_sync(self.__async_client.data_modeling.streams.create(items=items))
+
+    def list(self) -> StreamList:
+        """
+        `List streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/listStreams>`_.
+
+        Note:
+            There is no paging limit parameter: the endpoint returns all streams in the project
+            (projects are expected to have few streams).
+
+        Returns:
+            StreamList: The streams in the project.
+
+        Examples:
+
+            List all streams in the project:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.list()
+        """
+        return run_sync(self.__async_client.data_modeling.streams.list())
+
+    def retrieve(self, external_id: str, include_statistics: bool | None = None) -> Stream | None:
+        """
+        `Retrieve a stream <https://api-docs.cognite.com/20230101/tag/Streams/operation/getStream>`_.
+
+        Args:
+            external_id (str): External ID of the stream to retrieve.
+            include_statistics (bool | None): When ``True``, usage statistics will be returned together
+                with stream settings. Computing statistics can be expensive.
+
+        Returns:
+            Stream | None: The stream metadata (and optionally statistics), or ``None`` if not found.
+
+        Examples:
+
+            Retrieve a stream by external ID:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> res = client.data_modeling.streams.retrieve("my-stream")
+
+            Retrieve a stream with usage statistics:
+
+                >>> res = client.data_modeling.streams.retrieve(
+                ...     "my-stream",
+                ...     include_statistics=True,
+                ... )
+        """
+        return run_sync(
+            self.__async_client.data_modeling.streams.retrieve(
+                external_id=external_id, include_statistics=include_statistics
+            )
+        )
+
+    def delete(self, external_id: str | SequenceNotStr[str]) -> None:
+        """
+        `Delete streams <https://api-docs.cognite.com/20230101/tag/Streams/operation/deleteStreams>`_.
+
+        Note:
+            Deletion is a soft delete that retains capacity for an extended period;
+            prefer deleting only when necessary.
+
+        Args:
+            external_id (str | SequenceNotStr[str]): External ID or list of external IDs of
+                streams to delete.
+
+        Examples:
+
+            Delete a single stream:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> # async_client = AsyncCogniteClient()  # another option
+                >>> client.data_modeling.streams.delete("my-stream")
+
+            Delete multiple streams:
+
+                >>> client.data_modeling.streams.delete(["stream-a", "stream-b"])
+        """
+        return run_sync(self.__async_client.data_modeling.streams.delete(external_id=external_id))