diff --git a/src/pydase/utils/helpers.py b/src/pydase/utils/helpers.py index 2c0db14..4ac4d8e 100644 --- a/src/pydase/utils/helpers.py +++ b/src/pydase/utils/helpers.py @@ -8,6 +8,21 @@ from typing import Any logger = logging.getLogger(__name__) +def parse_serialized_key(serialized_key: str) -> str | int | float: + processed_key: int | float | str = serialized_key + if serialized_key.startswith("["): + assert serialized_key.endswith("]") + processed_key = serialized_key[1:-1] + if '"' in processed_key or "'" in processed_key: + processed_key = processed_key[1:-1] + elif "." in processed_key: + processed_key = float(processed_key) + else: + processed_key = int(processed_key) + + return processed_key + + def parse_full_access_path(path: str) -> list[str]: """ Splits a full access path into its atomic parts, separating attribute names, numeric @@ -77,6 +92,20 @@ def get_class_and_instance_attributes(obj: object) -> dict[str, Any]: return dict(chain(type(obj).__dict__.items(), obj.__dict__.items())) +def get_object_by_path_parts(target_obj: Any, path_parts: list[str]) -> Any: + for part in path_parts: + if part.startswith("["): + deserialized_part = parse_serialized_key(part) + target_obj = target_obj[deserialized_part] + else: + try: + target_obj = getattr(target_obj, part) + except AttributeError: + logger.debug("Attribute % does not exist in the object.", part) + return None + return target_obj + + def get_object_attr_from_path(target_obj: Any, path: str) -> Any: """ Traverse the object tree according to the given path. diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py index 80d4aba..8858a30 100644 --- a/tests/utils/test_helpers.py +++ b/tests/utils/test_helpers.py @@ -4,6 +4,7 @@ import pydase import pytest from pydase.utils.helpers import ( get_object_attr_from_path, + get_object_by_path_parts, get_path_from_path_parts, is_property_attribute, parse_full_access_path, @@ -53,6 +54,39 @@ def test_get_path_from_path_parts(path_parts: list[str], expected: str) -> None: assert get_path_from_path_parts(path_parts) == expected +class SubService(pydase.DataService): + name = "SubService" + some_int = 1 + some_float = 1.0 + + +class MyService(pydase.DataService): + def __init__(self) -> None: + super().__init__() + self.some_float = 1.0 + self.subservice = SubService() + self.list_attr = [1.0, SubService()] + self.dict_attr = {"foo": SubService()} + + +service_instance = MyService() + + +@pytest.mark.parametrize( + "path_parts, expected", + [ + (["some_float"], service_instance.some_float), + (["subservice"], service_instance.subservice), + (["list_attr", "[0]"], service_instance.list_attr[0]), + (["list_attr", "[1]"], service_instance.list_attr[1]), + (["dict_attr", '["foo"]'], service_instance.dict_attr["foo"]), + (["dict_attr", '["foo"]', "name"], service_instance.dict_attr["foo"].name), + ], +) +def test_get_object_by_path_parts(path_parts: list[str], expected: Any) -> None: + assert get_object_by_path_parts(service_instance, path_parts) == expected + + @pytest.mark.parametrize( "attr_name, expected", [