Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions cognite/client/_api/data_modeling/streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ 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>`_.

Note:
This method sends the alpha ``cdf-version`` header so that the alpha field
``settings.limits.max_unique_properties_with_data`` is returned, see
:class:`~cognite.client.data_classes.data_modeling.streams.StreamLimitSettings`.

Args:
items (StreamWrite | Sequence[StreamWrite]): One or more streams to create.

Expand Down Expand Up @@ -73,6 +78,7 @@ async def create(self, items: StreamWrite | Sequence[StreamWrite]) -> Stream | S
list_cls=StreamList,
resource_cls=Stream,
input_resource_cls=StreamWrite,
headers=self._alpha_version_header(),
)

async def list(self) -> StreamList:
Expand All @@ -82,6 +88,10 @@ async def list(self) -> StreamList:
There is no paging limit parameter: the endpoint returns all streams in the project
(projects are expected to have few streams).

This method sends the alpha ``cdf-version`` header so that the alpha field
``settings.limits.max_unique_properties_with_data`` is returned, see
:class:`~cognite.client.data_classes.data_modeling.streams.StreamLimitSettings`.

Returns:
StreamList: The streams in the project.

Expand All @@ -95,15 +105,23 @@ async def list(self) -> StreamList:
>>> res = client.data_modeling.streams.list()
"""
self._warning.warn()
return await self._list(method="GET", list_cls=StreamList, resource_cls=Stream)
return await self._list(
method="GET", list_cls=StreamList, resource_cls=Stream, headers=self._alpha_version_header()
)

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>`_.

Note:
This method sends the alpha ``cdf-version`` header so that the alpha field
``settings.limits.max_unique_properties_with_data`` is returned, see
:class:`~cognite.client.data_classes.data_modeling.streams.StreamLimitSettings`.

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.
with stream settings, i.e. the ``consumed`` values of the stream limits, including the number
of unique properties with data. Computing statistics can be expensive.

Returns:
Stream | None: The stream metadata (and optionally statistics), or ``None`` if not found.
Expand All @@ -129,6 +147,7 @@ async def retrieve(self, external_id: str, include_statistics: bool | None = Non
cls=Stream,
identifier=Identifier(external_id),
params={"includeStatistics": include_statistics} if include_statistics is not None else None,
headers=self._alpha_version_header(),
)

async def delete(self, external_id: str | SequenceNotStr[str]) -> None:
Expand Down
17 changes: 16 additions & 1 deletion cognite/client/_sync_api/data_modeling/streams.py

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

10 changes: 10 additions & 0 deletions cognite/client/data_classes/data_modeling/streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,17 +55,25 @@ def __init__(
max_records_total: StreamLimit,
max_giga_bytes_total: StreamLimit,
max_filtering_interval: str | None = None,
max_unique_properties_with_data: StreamLimit | None = None,
) -> None:
self.max_records_total = max_records_total
self.max_giga_bytes_total = max_giga_bytes_total
self.max_filtering_interval = max_filtering_interval
self.max_unique_properties_with_data = max_unique_properties_with_data

@classmethod
def _load(cls, resource: dict[str, Any]) -> Self:
max_unique_properties_with_data = resource.get("maxUniquePropertiesWithData")
return cls(
max_records_total=StreamLimit._load(resource["maxRecordsTotal"]),
max_giga_bytes_total=StreamLimit._load(resource["maxGigaBytesTotal"]),
max_filtering_interval=resource.get("maxFilteringInterval"),
max_unique_properties_with_data=(
StreamLimit._load(max_unique_properties_with_data)
if max_unique_properties_with_data is not None
else None
),
)

def dump(self, camel_case: bool = True) -> dict[str, Any]:
Expand All @@ -75,6 +83,8 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
}
if self.max_filtering_interval is not None:
out["max_filtering_interval"] = self.max_filtering_interval
if self.max_unique_properties_with_data is not None:
out["max_unique_properties_with_data"] = self.max_unique_properties_with_data.dump(camel_case=camel_case)
return convert_all_keys_to_camel_case(out) if camel_case else out


Expand Down
27 changes: 27 additions & 0 deletions tests/tests_unit/test_api/test_data_modeling/test_streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,10 @@ def test_list_parses_items(
out = cognite_client.data_modeling.streams.list()
assert isinstance(out, StreamList)
assert out[0].external_id == "st1"
assert out[0].settings.limits.max_unique_properties_with_data is None
assert out.dump() == stream_list_response["items"]
(request,) = httpx2_mock.get_requests()
assert request.headers["cdf-version"] == "20230101-alpha"

def test_templates_list_parses_items_and_sends_alpha_header(
self,
Expand Down Expand Up @@ -125,6 +129,28 @@ def test_retrieve_include_statistics_query(
requests = httpx2_mock.get_requests()
assert len(requests) == 1
assert requests[0].url.params["includeStatistics"].lower() == "true"
assert requests[0].headers["cdf-version"] == "20230101-alpha"

def test_retrieve_parses_max_unique_properties_with_data(
self,
cognite_client: CogniteClient,
httpx2_mock: HTTPXMock,
streams_base_url: str,
stream_response: dict,
) -> None:
stream_response["settings"]["limits"]["maxUniquePropertiesWithData"] = {"provisioned": 1000, "consumed": 42}
httpx2_mock.add_response(
method="GET",
url=re.compile(re.escape(streams_base_url) + r"/st1(?:\?.+)?$"),
json=stream_response,
)
out = cognite_client.data_modeling.streams.retrieve("st1", include_statistics=True)
assert isinstance(out, Stream)
unique_properties = out.settings.limits.max_unique_properties_with_data
assert unique_properties is not None
assert unique_properties.provisioned == 1000
assert unique_properties.consumed == 42
assert out.dump() == stream_response

def test_create_posts_single_item(
self,
Expand All @@ -145,6 +171,7 @@ def test_create_posts_single_item(
assert isinstance(out, Stream)
assert len(requests) == 1
assert requests[0].url.path.endswith("/streams")
assert requests[0].headers["cdf-version"] == "20230101-alpha"
assert jsgz_load(requests[0].content) == {
"items": [{"externalId": "st1", "settings": {"template": {"name": "ImmutableTestStream"}}}]
}
Expand Down
Loading