diff --git a/src/pydase/utils/serialization/serializer.py b/src/pydase/utils/serialization/serializer.py index b3d5028..bfc86ed 100644 --- a/src/pydase/utils/serialization/serializer.py +++ b/src/pydase/utils/serialization/serializer.py @@ -572,6 +572,66 @@ def generate_serialized_data_paths( return paths +def add_prefix_to_full_access_path( + serialized_obj: SerializedObject, prefix: str +) -> Any: + """Recursively adds a specified prefix to all full access paths of the serialized + object. + + Args: + data: + The serialized object to process. + prefix: + The prefix string to prepend to each full access path. + + Returns: + The modified serialized object with the prefix added to all full access paths. + + Example: + ```python + >>> data = { + ... "full_access_path": "some_path", + ... "value": { + ... "item": { + ... "full_access_path": "some_item_path", + ... "value": 1.0 + ... } + ... } + ... } + ... + ... modified_data = add_prefix_to_full_access_path(data, 'prefix.') + {"full_access_path": "prefix.some_path", "value": {"item": {"full_access_path": + "prefix.some_item_path", "value": 1.0}}} + ``` + """ + + try: + serialized_obj["full_access_path"] = prefix + serialized_obj["full_access_path"] + + if isinstance(serialized_obj["value"], list): + for value in serialized_obj["value"]: + value["full_access_path"] = prefix + value["full_access_path"] + add_prefix_to_full_access_path( + cast(SerializedObject, value["value"]), prefix + ) + + elif isinstance(serialized_obj["value"], dict): + for value in cast( + dict[str, SerializedObject], serialized_obj["value"] + ).values(): + value["full_access_path"] = prefix + value["full_access_path"] + add_prefix_to_full_access_path( + cast(SerializedObject, value["value"]), prefix + ) + except TypeError: + # passed object is not a dict (cannot use __get__ on it) + pass + except KeyError: + # passed dictionary is not a serialized object + pass + return serialized_obj + + def serialized_dict_is_nested_object(serialized_dict: SerializedObject) -> bool: value = serialized_dict["value"] # We are excluding Quantity here as the value corresponding to the "value" key is diff --git a/tests/utils/serialization/test_serializer.py b/tests/utils/serialization/test_serializer.py index 6c1ab81..341db12 100644 --- a/tests/utils/serialization/test_serializer.py +++ b/tests/utils/serialization/test_serializer.py @@ -12,6 +12,7 @@ from pydase.utils.decorators import frontend from pydase.utils.serialization.serializer import ( SerializationPathError, SerializedObject, + add_prefix_to_full_access_path, dump, generate_serialized_data_paths, get_container_item_by_key, @@ -1070,3 +1071,109 @@ def test_get_data_paths_from_serialized_object(obj: Any, expected: list[str]) -> ) def test_generate_serialized_data_paths(obj: Any, expected: list[str]) -> None: assert generate_serialized_data_paths(dump(obj=obj)["value"]) == expected + + +@pytest.mark.parametrize( + "serialized_obj, prefix, expected", + [ + ( + { + "full_access_path": "new_attr", + "value": { + "name": { + "full_access_path": "new_attr.name", + "value": "MyService", + } + }, + }, + "prefix.", + { + "full_access_path": "prefix.new_attr", + "value": { + "name": { + "full_access_path": "prefix.new_attr.name", + "value": "MyService", + } + }, + }, + ), + ( + { + "full_access_path": "new_attr", + "value": [ + { + "full_access_path": "new_attr[0]", + "value": 1.0, + } + ], + }, + "prefix.", + { + "full_access_path": "prefix.new_attr", + "value": [ + { + "full_access_path": "prefix.new_attr[0]", + "value": 1.0, + } + ], + }, + ), + ( + { + "full_access_path": "new_attr", + "value": { + "key": { + "full_access_path": 'new_attr["key"]', + "value": 1.0, + } + }, + }, + "prefix.", + { + "full_access_path": "prefix.new_attr", + "value": { + "key": { + "full_access_path": 'prefix.new_attr["key"]', + "value": 1.0, + } + }, + }, + ), + ( + { + "full_access_path": "new_attr", + "value": {"magnitude": 10, "unit": "meter"}, + }, + "prefix.", + { + "full_access_path": "prefix.new_attr", + "value": {"magnitude": 10, "unit": "meter"}, + }, + ), + ( + { + "full_access_path": "quantity_list", + "value": [ + { + "full_access_path": "quantity_list[0]", + "value": {"magnitude": 1.0, "unit": "A"}, + } + ], + }, + "prefix.", + { + "full_access_path": "prefix.quantity_list", + "value": [ + { + "full_access_path": "prefix.quantity_list[0]", + "value": {"magnitude": 1.0, "unit": "A"}, + } + ], + }, + ), + ], +) +def test_add_prefix_to_full_access_path( + serialized_obj: SerializedObject, prefix: str, expected: SerializedObject +) -> None: + assert add_prefix_to_full_access_path(serialized_obj, prefix) == expected