KeyError on retrieving an ExtractionPipeline with partial contact information

CHANGELOG.md

@@ -17,6 +17,12 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [7.70.2] - 2024-12-04
+### Fixed
+- Retrieving `ExtractionPipeline` either with `client.extraction_pipelines.retrieve` or 
+  `client.extraction_pipelines.list` no longer raises a `KeyError` if any of the pipline properties have a contact
+  with a `None` value.
+
 ## [7.70.1] - 2024-12-04
 ### Fixed
 - Fix `workflows.executions.retrieve_detailed` type for `SimulationInputOverride` to allow for `None` value for `unit`.

cognite/client/_version.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
 
-__version__ = "7.70.1"
+__version__ = "7.70.2"
 
 __api_subversion__ = "20230101"

cognite/client/data_classes/extractionpipelines.py

@@ -30,27 +30,24 @@ class ExtractionPipelineContact(CogniteObject):
     """A contact for an extraction pipeline
 
     Args:
-        name (str): Name of contact
-        email (str): Email address of contact
-        role (str): Role of contact, such as Owner, Maintainer, etc.
-        send_notification (bool): Whether to send notifications to this contact or not
+        name (str | None): Name of contact
+        email (str | None): Email address of contact
+        role (str | None): Role of contact, such as Owner, Maintainer, etc.
+        send_notification (bool | None): Whether to send notifications to this contact or not
     """
 
-    def __init__(self, name: str, email: str, role: str, send_notification: bool) -> None:
+    def __init__(
+        self,
+        name: str | None = None,
+        email: str | None = None,
+        role: str | None = None,
+        send_notification: bool | None = None,
+    ) -> None:
         self.name = name
         self.email = email
         self.role = role
         self.send_notification = send_notification
 
-    @classmethod
-    def _load(cls, resource: dict, cognite_client: CogniteClient | None = None) -> ExtractionPipelineContact:
-        return cls(
-            name=resource["name"],
-            email=resource["email"],
-            role=resource["role"],
-            send_notification=resource["sendNotification"],
-        )
-
 
 @dataclass
 class ExtractionPipelineNotificationConfiguration(CogniteObject):

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "cognite-sdk"
 
-version = "7.70.1"
+version = "7.70.2"
 
 description = "Cognite Python SDK"
 readme = "README.md"

tests/tests_unit/test_data_classes/test_extraction_pipipelines.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+import pytest
+
+from cognite.client.data_classes.extractionpipelines import ExtractionPipelineContact
+
+
+class TestExtractionPipeline:
+    @pytest.mark.parametrize(
+        "raw, expected",
+        [
+            pytest.param(
+                {"name": "Homer", "email": "[email protected]"},
+                ExtractionPipelineContact(name="Homer", email="[email protected]"),
+                id="Partial Contact",
+            ),
+            pytest.param(
+                {"name": "Marge", "email": "[email protected]", "role": "Wife", "sendNotification": True},
+                ExtractionPipelineContact(
+                    name="Marge", email="[email protected]", role="Wife", send_notification=True
+                ),
+                id="Full Contact",
+            ),
+        ],
+    )
+    def test_load_dump_contracts(self, raw: dict[str, Any], expected: ExtractionPipelineContact) -> None:
+        loaded = ExtractionPipelineContact._load(raw)
+        assert loaded == expected
+        assert raw == loaded.dump()