mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-20 08:20:02 +02:00
adds parse_serialized_key and get_object_by_path_parts helper methods
This commit is contained in:
parent
b29c86ac2c
commit
2898b62b9c
@ -8,6 +8,21 @@ from typing import Any
|
|||||||
logger = logging.getLogger(__name__)
|
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]:
|
def parse_full_access_path(path: str) -> list[str]:
|
||||||
"""
|
"""
|
||||||
Splits a full access path into its atomic parts, separating attribute names, numeric
|
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()))
|
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:
|
def get_object_attr_from_path(target_obj: Any, path: str) -> Any:
|
||||||
"""
|
"""
|
||||||
Traverse the object tree according to the given path.
|
Traverse the object tree according to the given path.
|
||||||
|
@ -4,6 +4,7 @@ import pydase
|
|||||||
import pytest
|
import pytest
|
||||||
from pydase.utils.helpers import (
|
from pydase.utils.helpers import (
|
||||||
get_object_attr_from_path,
|
get_object_attr_from_path,
|
||||||
|
get_object_by_path_parts,
|
||||||
get_path_from_path_parts,
|
get_path_from_path_parts,
|
||||||
is_property_attribute,
|
is_property_attribute,
|
||||||
parse_full_access_path,
|
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
|
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(
|
@pytest.mark.parametrize(
|
||||||
"attr_name, expected",
|
"attr_name, expected",
|
||||||
[
|
[
|
||||||
|
Loading…
x
Reference in New Issue
Block a user