Tolerate extra keys when deserializing HTTP response

geti_sdk/utils/serialization_helpers.py

@@ -14,6 +14,7 @@
 
 from typing import Any, Dict, Optional, Type, TypeVar, cast
 
+from attr import fields, has
 from omegaconf import OmegaConf
 from omegaconf.errors import ConfigKeyError, ConfigTypeError, MissingMandatoryValue
 
@@ -32,15 +33,32 @@ def deserialize_dictionary(
     :return: Object of type `output_type`, holding the data passed in
         `input_dictionary`.
     """
-    model_dict_config = OmegaConf.create(input_dictionary)
+
+    def prune_dict(data: dict, cls: Type[Any]) -> dict:
+        """Recursively prune a dictionary to match the structure of an attr class."""
+        pruned_data = {}
+        for attribute in fields(cls):
+            key = attribute.name
+            if key in data:
+                value = data[key]
+                # Check if the field is itself a structured class
+                if has(attribute.type) and isinstance(value, dict):
+                    # Recursively prune the nested dictionary
+                    pruned_data[key] = prune_dict(value, attribute.type)
+                else:
+                    pruned_data[key] = value
+        return pruned_data
+
+    filtered_input_dictionary = prune_dict(input_dictionary, output_type)
+    model_dict_config = OmegaConf.create(filtered_input_dictionary)
     schema = OmegaConf.structured(output_type)
     schema_error: Optional[DataModelMismatchException] = None
     try:
         values = OmegaConf.merge(schema, model_dict_config)
         output = cast(output_type, OmegaConf.to_object(values))
     except (ConfigKeyError, MissingMandatoryValue, ConfigTypeError) as error:
         schema_error = DataModelMismatchException(
-            input_dictionary=input_dictionary,
+            input_dictionary=filtered_input_dictionary,
             output_data_model=output_type,
             message=error.args[0],
             error_type=type(error),

tests/pre-merge/unit/utils/test_utils_unit.py

@@ -30,9 +30,8 @@ def test_deserialize_dictionary(self, fxt_project_dictionary: dict):
         """
         Verifies that deserializing a dictionary to a python object works.
 
-        Also tests that a DataModelMismatchException is raised in case:
-            1. the input dictionary contains an invalid key
-            2. the input dictionary misses a required key
+        The test checks that a DataModelMismatchException is raised in case of a missing key.
+        It also verifies that the presence of additional keys in the input dictionary is not a problem.
         """
 
         # Arrange
@@ -41,6 +40,11 @@ def test_deserialize_dictionary(self, fxt_project_dictionary: dict):
         dictionary_with_extra_key = copy.deepcopy(fxt_project_dictionary)
         dictionary_with_extra_key.update({"invalid_key": "invalidness"})
 
+        dictionary_with_nested_extra_key = copy.deepcopy(fxt_project_dictionary)
+        dictionary_with_nested_extra_key["pipeline"].update(
+            {"invalid_key": "invalidness"}
+        )
+
         dictionary_with_missing_key = copy.deepcopy(fxt_project_dictionary)
         dictionary_with_missing_key.pop("pipeline")
 
@@ -54,15 +58,19 @@ def test_deserialize_dictionary(self, fxt_project_dictionary: dict):
         assert project.get_trainable_tasks()[0].type == TaskType.DETECTION
 
         # Act and assert
-        with pytest.raises(DataModelMismatchException):
-            deserialize_dictionary(
-                input_dictionary=dictionary_with_extra_key, output_type=object_type
-            )
+        deserialize_dictionary(
+            input_dictionary=dictionary_with_extra_key, output_type=object_type
+        )
+
+        # Act and assert
+        deserialize_dictionary(
+            input_dictionary=dictionary_with_nested_extra_key, output_type=object_type
+        )
 
         # Act and assert
         with pytest.raises(DataModelMismatchException):
             deserialize_dictionary(
-                input_dictionary=dictionary_with_extra_key, output_type=object_type
+                input_dictionary=dictionary_with_missing_key, output_type=object_type
             )
 
     def test_generate_segmentation_labels(self):