From fa12c88b2f96ccc6aa21fa8dd390ed9a8091c3f6 Mon Sep 17 00:00:00 2001 From: Ankit Agarwal Date: Mon, 5 Oct 2026 13:48:53 +0530 Subject: [PATCH 1/3] Adding data domain external id in the transformation apis --- .../client/_api/transformations/__init__.py | 6 ++ .../_sync_api/transformations/__init__.py | 3 + .../data_classes/transformations/__init__.py | 19 ++++ .../test_api/test_transformations.py | 49 ++++++++++ .../test_transformation.py | 90 +++++++++++++++++++ 5 files changed, 167 insertions(+) create mode 100644 tests/tests_unit/test_data_classes/test_transformations/test_transformation.py diff --git a/cognite/client/_api/transformations/__init__.py b/cognite/client/_api/transformations/__init__.py index 6c517a747d..17cd1ea389 100644 --- a/cognite/client/_api/transformations/__init__.py +++ b/cognite/client/_api/transformations/__init__.py @@ -300,6 +300,7 @@ async def list( data_set_external_ids: str | list[str] | None = None, tags: TagsFilter | None = None, limit: int | None = DEFAULT_LIMIT_READ, + data_domain_external_ids: str | list[str] | None = None, ) -> TransformationList: """`List all transformations `_. @@ -317,6 +318,7 @@ async def list( data_set_external_ids (str | list[str] | None): Return only transformations in the specified data sets with these external id(s). tags (TagsFilter | None): Return only the resource matching the specified tags constraints. It only supports ContainsAny as of now. limit (int | None): Limits the number of results to be returned. To retrieve all results use limit=-1, default limit is 25. + data_domain_external_ids (str | list[str] | None): Return only transformations belonging to one of the specified data domain(s). Returns: TransformationList: List of transformations @@ -331,6 +333,9 @@ async def list( >>> transformations_list = client.transformations.list() """ ds_ids = IdentifierSequence.load(data_set_ids, data_set_external_ids, id_name="data_set").as_dicts() + domain_external_ids = ( + [data_domain_external_ids] if isinstance(data_domain_external_ids, str) else data_domain_external_ids + ) or None filter = TransformationFilter( include_public=include_public, @@ -344,6 +349,7 @@ async def list( last_updated_time=last_updated_time, tags=tags, data_set_ids=ds_ids or None, + data_domain_external_ids=domain_external_ids, ).dump(camel_case=True) return await self._list( diff --git a/cognite/client/_sync_api/transformations/__init__.py b/cognite/client/_sync_api/transformations/__init__.py index a6ab19fd6f..f910450b35 100644 --- a/cognite/client/_sync_api/transformations/__init__.py +++ b/cognite/client/_sync_api/transformations/__init__.py @@ -270,6 +270,7 @@ def list( data_set_external_ids: str | list[str] | None = None, tags: TagsFilter | None = None, limit: int | None = DEFAULT_LIMIT_READ, + data_domain_external_ids: str | list[str] | None = None, ) -> TransformationList: """ `List all transformations `_. @@ -288,6 +289,7 @@ def list( data_set_external_ids (str | list[str] | None): Return only transformations in the specified data sets with these external id(s). tags (TagsFilter | None): Return only the resource matching the specified tags constraints. It only supports ContainsAny as of now. limit (int | None): Limits the number of results to be returned. To retrieve all results use limit=-1, default limit is 25. + data_domain_external_ids (str | list[str] | None): Return only transformations belonging to one of the specified data domain(s). Returns: TransformationList: List of transformations @@ -316,6 +318,7 @@ def list( data_set_external_ids=data_set_external_ids, tags=tags, limit=limit, + data_domain_external_ids=data_domain_external_ids, ) ) diff --git a/cognite/client/data_classes/transformations/__init__.py b/cognite/client/data_classes/transformations/__init__.py index 3b39a3167b..462d217709 100644 --- a/cognite/client/data_classes/transformations/__init__.py +++ b/cognite/client/data_classes/transformations/__init__.py @@ -217,6 +217,7 @@ class Transformation(WriteableCogniteResourceWithClientRef["TransformationWrite" source_session (SessionDetails | None): Details for the session used to read from the source project. destination_session (SessionDetails | None): Details for the session used to write to the destination project. tags (list[str] | None): No description. + data_domain_external_id (str | None): The external ID of the data domain this transformation belongs to. """ def __init__( @@ -245,6 +246,7 @@ def __init__( source_session: SessionDetails | None = None, destination_session: SessionDetails | None = None, tags: list[str] | None = None, + data_domain_external_id: str | None = None, ) -> None: self.external_id = external_id self.name = name @@ -254,6 +256,7 @@ def __init__( self.source_nonce = source_nonce self.destination_nonce = destination_nonce self.data_set_id = data_set_id + self.data_domain_external_id = data_domain_external_id self.id = id self.query = query self.destination = destination @@ -300,6 +303,7 @@ def as_write(self) -> TransformationWrite: source_oidc_credentials=self.source_oidc_credentials, destination_oidc_credentials=self.destination_oidc_credentials, data_set_id=self.data_set_id, + data_domain_external_id=self.data_domain_external_id, source_nonce=self.source_nonce, destination_nonce=self.destination_nonce, tags=self.tags, @@ -334,6 +338,7 @@ def copy(self) -> Transformation: blocked=self.blocked, schedule=self.schedule, data_set_id=self.data_set_id, + data_domain_external_id=self.data_domain_external_id, source_nonce=self.source_nonce, destination_nonce=self.destination_nonce, source_session=self.source_session, @@ -412,6 +417,7 @@ def _load(cls, resource: dict) -> Transformation: blocked=TransformationBlockedInfo._load_if(resource.get("blocked")), schedule=TransformationSchedule._load_if(resource.get("schedule")), data_set_id=resource.get("dataSetId"), + data_domain_external_id=resource.get("dataDomainExternalId"), source_nonce=NonceCredentials._load_if(resource.get("sourceNonce")), destination_nonce=NonceCredentials._load_if(resource.get("destinationNonce")), source_session=SessionDetails._load_if(resource.get("sourceSession")), @@ -440,6 +446,7 @@ class TransformationWrite(WriteableCogniteResource["TransformationWrite"], _Tran source_nonce (NonceCredentials | None): Single use credentials to bind to a CDF session for reading. destination_nonce (NonceCredentials | None): Single use credentials to bind to a CDF session for writing. tags (list[str] | None): No description. + data_domain_external_id (str | None): The external ID of the data domain this transformation belongs to. """ def __init__( @@ -457,6 +464,7 @@ def __init__( source_nonce: NonceCredentials | None = None, destination_nonce: NonceCredentials | None = None, tags: list[str] | None = None, + data_domain_external_id: str | None = None, ) -> None: self.external_id = external_id self.name = name @@ -466,6 +474,7 @@ def __init__( self.source_nonce = source_nonce self.destination_nonce = destination_nonce self.data_set_id = data_set_id + self.data_domain_external_id = data_domain_external_id self.query = query self.destination = destination self.conflict_mode = conflict_mode @@ -485,6 +494,7 @@ def _load(cls, resource: dict) -> TransformationWrite: source_oidc_credentials=OidcCredentials._load_if(resource.get("sourceOidcCredentials")), destination_oidc_credentials=OidcCredentials._load_if(resource.get("destinationOidcCredentials")), data_set_id=resource.get("dataSetId"), + data_domain_external_id=resource.get("dataDomainExternalId"), source_nonce=NonceCredentials._load_if(resource.get("sourceNonce")), destination_nonce=NonceCredentials._load_if(resource.get("destinationNonce")), tags=resource.get("tags"), @@ -505,6 +515,7 @@ def copy(self) -> TransformationWrite: self.source_nonce, self.destination_nonce, self.tags, + self.data_domain_external_id, ) def dump(self, camel_case: bool = True) -> dict[str, Any]: @@ -598,6 +609,10 @@ def ignore_null_fields(self) -> _PrimitiveTransformationUpdate: def data_set_id(self) -> _PrimitiveTransformationUpdate: return TransformationUpdate._PrimitiveTransformationUpdate(self, "dataSetId") + @property + def data_domain_external_id(self) -> _PrimitiveTransformationUpdate: + return TransformationUpdate._PrimitiveTransformationUpdate(self, "dataDomainExternalId") + @property def tags(self) -> _ListTransformationUpdate: return TransformationUpdate._ListTransformationUpdate(self, "tags") @@ -626,6 +641,7 @@ def _get_update_properties(cls, item: CogniteResource | None = None) -> list[Pro PropertySpec("is_public", is_nullable=False), PropertySpec("ignore_null_fields", is_nullable=False), PropertySpec("data_set_id"), + PropertySpec("data_domain_external_id"), PropertySpec("tags", is_list=True), ] @@ -683,6 +699,7 @@ class TransformationFilter(CogniteFilter): last_updated_time (dict[str, Any] | TimestampRange | None): Range between two timestamps data_set_ids (list[dict[str, Any]] | None): Return only transformations in the specified data sets with these ids, e.g. [{"id": 1}, {"externalId": "foo"}]. tags (TagsFilter | None): Return only the resource matching the specified tags constraints. It only supports ContainsAny as of now. + data_domain_external_ids (list[str] | None): Return only transformations belonging to one of the specified data domains, e.g. ["my-domain"]. """ def __init__( @@ -698,6 +715,7 @@ def __init__( last_updated_time: dict[str, Any] | TimestampRange | None = None, data_set_ids: list[dict[str, Any]] | None = None, tags: TagsFilter | None = None, + data_domain_external_ids: list[str] | None = None, ) -> None: self.include_public = include_public self.name_regex = name_regex @@ -710,6 +728,7 @@ def __init__( self.last_updated_time = last_updated_time self.data_set_ids = data_set_ids self.tags = tags + self.data_domain_external_ids = data_domain_external_ids def dump(self, camel_case: bool = True) -> dict[str, Any]: obj = super().dump(camel_case=camel_case) diff --git a/tests/tests_unit/test_api/test_transformations.py b/tests/tests_unit/test_api/test_transformations.py index d721651022..04a6edeb60 100644 --- a/tests/tests_unit/test_api/test_transformations.py +++ b/tests/tests_unit/test_api/test_transformations.py @@ -1,10 +1,18 @@ from __future__ import annotations +import re +from typing import TYPE_CHECKING + import pytest +from pytest_httpx2 import HTTPXMock from cognite.client.credentials import OAuthClientCredentials from cognite.client.data_classes.iam import ClientCredentials from cognite.client.data_classes.transformations.common import OidcCredentials +from tests.utils import get_url, jsgz_load + +if TYPE_CHECKING: + from cognite.client import AsyncCogniteClient, CogniteClient @pytest.fixture @@ -49,3 +57,44 @@ def test_oidc_credentials_as_client_credentials(oidc_credentials: OidcCredential assert isinstance(client_creds, ClientCredentials) assert client_creds == ClientCredentials("id", "secret") + + +@pytest.fixture +def mock_transformations_list_response(httpx2_mock: HTTPXMock, async_client: AsyncCogniteClient) -> HTTPXMock: + url_pattern = re.compile(re.escape(get_url(async_client.transformations)) + r"/transformations/filter(?:\?.+)?$") + httpx2_mock.add_response(method="POST", url=url_pattern, status_code=200, json={"items": []}) + return httpx2_mock + + +class TestTransformationsListDataDomainExternalIdsFilter: + def test_omitted_when_not_given( + self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock + ) -> None: + cognite_client.transformations.list() + + sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] + assert "dataDomainExternalIds" not in sent_filter + + def test_omitted_when_empty_list_given( + self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock + ) -> None: + cognite_client.transformations.list(data_domain_external_ids=[]) + + sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] + assert "dataDomainExternalIds" not in sent_filter + + def test_single_string_is_wrapped_in_list( + self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock + ) -> None: + cognite_client.transformations.list(data_domain_external_ids="my-domain") + + sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] + assert sent_filter["dataDomainExternalIds"] == ["my-domain"] + + def test_list_of_strings_is_passed_through( + self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock + ) -> None: + cognite_client.transformations.list(data_domain_external_ids=["domain-a", "domain-b"]) + + sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] + assert sent_filter["dataDomainExternalIds"] == ["domain-a", "domain-b"] diff --git a/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py new file mode 100644 index 0000000000..33806369f4 --- /dev/null +++ b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from cognite.client.data_classes.transformations import ( + Transformation, + TransformationFilter, + TransformationUpdate, + TransformationWrite, +) + + +@pytest.fixture +def transformation_resource() -> dict[str, Any]: + return { + "id": 1, + "externalId": "my-transformation", + "name": "My transformation", + "query": "SELECT 1", + "destination": {"type": "assets"}, + "conflictMode": "upsert", + "isPublic": True, + "ignoreNullFields": False, + "createdTime": 1, + "lastUpdatedTime": 2, + "owner": "someone", + "ownerIsCurrentUser": True, + "dataSetId": 123, + "dataDomainExternalId": "my-domain", + } + + +class TestTransformationDataDomainExternalId: + def test_load_sets_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: + transformation = Transformation._load(transformation_resource) + + assert transformation.data_domain_external_id == "my-domain" + + def test_load_defaults_to_none_when_absent(self, transformation_resource: dict[str, Any]) -> None: + del transformation_resource["dataDomainExternalId"] + + transformation = Transformation._load(transformation_resource) + + assert transformation.data_domain_external_id is None + + def test_dump_round_trip(self, transformation_resource: dict[str, Any]) -> None: + transformation = Transformation._load(transformation_resource) + + dumped = transformation.dump(camel_case=True) + + assert dumped["dataDomainExternalId"] == "my-domain" + + def test_as_write_preserves_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: + transformation = Transformation._load(transformation_resource) + + written = transformation.as_write() + + assert isinstance(written, TransformationWrite) + assert written.data_domain_external_id == "my-domain" + + def test_copy_preserves_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: + transformation = Transformation._load(transformation_resource) + + copied = transformation.copy() + + assert copied.data_domain_external_id == "my-domain" + + def test_transformation_write_dump_load_round_trip(self) -> None: + write = TransformationWrite(external_id="my-transformation", name="My transformation") + write.data_domain_external_id = "my-domain" + + loaded = TransformationWrite._load(write.dump(camel_case=True)) + + assert loaded.data_domain_external_id == "my-domain" + + def test_update_setter_dumps_expected_shape(self) -> None: + update = TransformationUpdate(id=1).data_domain_external_id.set("my-domain") + + dumped = update.dump(camel_case=True) + + assert dumped["update"]["dataDomainExternalId"] == {"set": "my-domain"} + + def test_filter_dumps_data_domain_external_ids(self) -> None: + filter = TransformationFilter(data_domain_external_ids=["my-domain", "other-domain"]) + + dumped = filter.dump(camel_case=True) + + assert dumped["dataDomainExternalIds"] == ["my-domain", "other-domain"] From 232d99bffb7c82b8c1aeb9e515b20914cdd77922 Mon Sep 17 00:00:00 2001 From: Ankit Agarwal Date: Mon, 12 Oct 2026 08:41:52 +0530 Subject: [PATCH 2/3] adding test cases --- .../test_transformation.py | 35 ------------------- 1 file changed, 35 deletions(-) diff --git a/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py index 33806369f4..e700a6acc2 100644 --- a/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py +++ b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py @@ -7,7 +7,6 @@ from cognite.client.data_classes.transformations import ( Transformation, TransformationFilter, - TransformationUpdate, TransformationWrite, ) @@ -33,25 +32,6 @@ def transformation_resource() -> dict[str, Any]: class TestTransformationDataDomainExternalId: - def test_load_sets_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: - transformation = Transformation._load(transformation_resource) - - assert transformation.data_domain_external_id == "my-domain" - - def test_load_defaults_to_none_when_absent(self, transformation_resource: dict[str, Any]) -> None: - del transformation_resource["dataDomainExternalId"] - - transformation = Transformation._load(transformation_resource) - - assert transformation.data_domain_external_id is None - - def test_dump_round_trip(self, transformation_resource: dict[str, Any]) -> None: - transformation = Transformation._load(transformation_resource) - - dumped = transformation.dump(camel_case=True) - - assert dumped["dataDomainExternalId"] == "my-domain" - def test_as_write_preserves_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: transformation = Transformation._load(transformation_resource) @@ -67,21 +47,6 @@ def test_copy_preserves_data_domain_external_id(self, transformation_resource: d assert copied.data_domain_external_id == "my-domain" - def test_transformation_write_dump_load_round_trip(self) -> None: - write = TransformationWrite(external_id="my-transformation", name="My transformation") - write.data_domain_external_id = "my-domain" - - loaded = TransformationWrite._load(write.dump(camel_case=True)) - - assert loaded.data_domain_external_id == "my-domain" - - def test_update_setter_dumps_expected_shape(self) -> None: - update = TransformationUpdate(id=1).data_domain_external_id.set("my-domain") - - dumped = update.dump(camel_case=True) - - assert dumped["update"]["dataDomainExternalId"] == {"set": "my-domain"} - def test_filter_dumps_data_domain_external_ids(self) -> None: filter = TransformationFilter(data_domain_external_ids=["my-domain", "other-domain"]) From dae672139bcfa6dac45b254fdb12161d128b6a47 Mon Sep 17 00:00:00 2001 From: Ankit Agarwal Date: Thu, 15 Oct 2026 15:47:03 +0530 Subject: [PATCH 3/3] fixing test cases Co-Authored-By: Claude Sonnet 5.5 --- .../test_api/test_transformations.py | 51 ++++++---------- .../test_transformation.py | 60 +++++++------------ 2 files changed, 39 insertions(+), 72 deletions(-) diff --git a/tests/tests_unit/test_api/test_transformations.py b/tests/tests_unit/test_api/test_transformations.py index 04a6edeb60..4d1365ff79 100644 --- a/tests/tests_unit/test_api/test_transformations.py +++ b/tests/tests_unit/test_api/test_transformations.py @@ -66,35 +66,22 @@ def mock_transformations_list_response(httpx2_mock: HTTPXMock, async_client: Asy return httpx2_mock -class TestTransformationsListDataDomainExternalIdsFilter: - def test_omitted_when_not_given( - self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock - ) -> None: - cognite_client.transformations.list() - - sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] - assert "dataDomainExternalIds" not in sent_filter - - def test_omitted_when_empty_list_given( - self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock - ) -> None: - cognite_client.transformations.list(data_domain_external_ids=[]) - - sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] - assert "dataDomainExternalIds" not in sent_filter - - def test_single_string_is_wrapped_in_list( - self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock - ) -> None: - cognite_client.transformations.list(data_domain_external_ids="my-domain") - - sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] - assert sent_filter["dataDomainExternalIds"] == ["my-domain"] - - def test_list_of_strings_is_passed_through( - self, cognite_client: CogniteClient, mock_transformations_list_response: HTTPXMock - ) -> None: - cognite_client.transformations.list(data_domain_external_ids=["domain-a", "domain-b"]) - - sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] - assert sent_filter["dataDomainExternalIds"] == ["domain-a", "domain-b"] +@pytest.mark.parametrize( + "data_domain_external_ids, expected", + [ + pytest.param(None, None, id="omitted when not given"), + pytest.param([], None, id="omitted when empty list given"), + pytest.param("my-domain", ["my-domain"], id="single string is wrapped in list"), + pytest.param(["domain-a", "domain-b"], ["domain-a", "domain-b"], id="list of strings passed through"), + ], +) +def test_list_data_domain_external_ids_filter( + cognite_client: CogniteClient, + mock_transformations_list_response: HTTPXMock, + data_domain_external_ids: str | list[str] | None, + expected: list[str] | None, +) -> None: + cognite_client.transformations.list(data_domain_external_ids=data_domain_external_ids) + + sent_filter = jsgz_load(mock_transformations_list_response.get_requests()[0].content)["filter"] + assert sent_filter.get("dataDomainExternalIds") == expected diff --git a/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py index e700a6acc2..c354f3826a 100644 --- a/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py +++ b/tests/tests_unit/test_data_classes/test_transformations/test_transformation.py @@ -1,55 +1,35 @@ from __future__ import annotations -from typing import Any - import pytest from cognite.client.data_classes.transformations import ( Transformation, - TransformationFilter, + TransformationUpdate, TransformationWrite, ) +from tests.utils import FakeCogniteResourceGenerator -@pytest.fixture -def transformation_resource() -> dict[str, Any]: - return { - "id": 1, - "externalId": "my-transformation", - "name": "My transformation", - "query": "SELECT 1", - "destination": {"type": "assets"}, - "conflictMode": "upsert", - "isPublic": True, - "ignoreNullFields": False, - "createdTime": 1, - "lastUpdatedTime": 2, - "owner": "someone", - "ownerIsCurrentUser": True, - "dataSetId": 123, - "dataDomainExternalId": "my-domain", - } - - -class TestTransformationDataDomainExternalId: - def test_as_write_preserves_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: - transformation = Transformation._load(transformation_resource) - - written = transformation.as_write() +@pytest.mark.dsl # requirement when using FakeCogniteResourceGenerator +@pytest.mark.parametrize("transformation_cls", (Transformation, TransformationWrite)) +def test_copy_preserves_all_fields(transformation_cls: type[Transformation] | type[TransformationWrite]) -> None: + transformation = FakeCogniteResourceGenerator().create_instance(transformation_cls) - assert isinstance(written, TransformationWrite) - assert written.data_domain_external_id == "my-domain" + copied = transformation.copy() - def test_copy_preserves_data_domain_external_id(self, transformation_resource: dict[str, Any]) -> None: - transformation = Transformation._load(transformation_resource) + assert copied is not transformation + assert type(copied) is transformation_cls + assert copied.dump() == transformation.dump() - copied = transformation.copy() - assert copied.data_domain_external_id == "my-domain" - - def test_filter_dumps_data_domain_external_ids(self) -> None: - filter = TransformationFilter(data_domain_external_ids=["my-domain", "other-domain"]) - - dumped = filter.dump(camel_case=True) +@pytest.mark.parametrize( + "value, expected", + [ + pytest.param("my-domain", {"set": "my-domain"}, id="set value"), + pytest.param(None, {"setNull": True}, id="set to None clears the value"), + ], +) +def test_update_data_domain_external_id(value: str | None, expected: dict[str, object]) -> None: + update = TransformationUpdate(external_id="my-transformation").data_domain_external_id.set(value) - assert dumped["dataDomainExternalIds"] == ["my-domain", "other-domain"] + assert update.dump()["update"] == {"dataDomainExternalId": expected}