Add sdbus.utils.parse_get_managed_objects

igo95862 · igo95862 · commit 87a56f86ee3a · 2024-03-10T21:58:31.000+05:00
Parses data from ObjectsManager's `get_managed_objects` calls.
It is similar to existing `parse_interfaces_added` function.

Also reduce code duplication in sdbus.utils using extra functions.
diff --git a/docs/utils.rst b/docs/utils.rst
@@ -59,3 +59,22 @@ Parsing utilities
             of interface class.
     :rtype: Tuple[str, Optional[Type[DbusInterfaceBaseAsync]]]
     :returns: Path of removed object and object's class (or ``None``).
+
+.. py:function:: parse_get_managed_objects(interfaces, managed_objects_data, on_unknown_interface='error', on_unknown_member='error')
+
+    Parse data from :py:meth:`get_managed_objects <sdbus.DbusObjectManagerInterfaceAsync.get_managed_objects>` call.
+
+    Takes an iterable of D-Bus interface classes (or a single class) and the method returned data.
+    Returns a dictionary where keys a paths of the managed objects and value is a tuple of class of the object
+    and dictionary of its python named properties and their values.
+
+    :param Iterable[DbusInterfaceBaseAsync] interfaces: Possible interfaces of the managed objects.
+        Can accept classes with multiple interfaces defined.
+    :param Dict interfaces_added_data: Data returned by ``get_managed_objects`` call.
+    :param str on_unknown_member: If an unknown D-Bus interface was encountered
+            either raise an ``"error"`` (default) or return ``"none"`` instead
+            of interface class.
+    :rtype: Dict[str, Tuple[Optional[Type[DbusInterfaceBaseAsync], Dict[str, Any]]]]
+    :returns: Dictionary where keys are paths and values are tuples of managed objects classes and their properties data.
+
+    *New in version 0.12.0.*
diff --git a/src/sdbus/utils.py b/src/sdbus/utils.py
@@ -44,11 +44,27 @@
 
     from .dbus_proxy_async_interfaces import DBUS_PROPERTIES_CHANGED_TYPING
 
+    InterfacesInputElements = Union[
+        DbusInterfaceBaseAsync,
+        Type[DbusInterfaceBaseAsync],
+    ]
+    InterfacesInput = Union[
+        InterfacesInputElements,
+        Iterable[InterfacesInputElements],
+    ]
+    InterfacesToClassMap = Dict[FrozenSet[str], Type[DbusInterfaceBaseAsync]]
+    OnUnknownMember = Literal['error', 'ignore', 'reuse']
+    OnUnknownInterface = Literal['error', 'none']
+    ParseGetManaged = Dict[
+        str,
+        Tuple[Optional[Type[DbusInterfaceBaseAsync]], Dict[str, Any]],
+    ]
+
 
 def parse_properties_changed(
-        interface: Union[DbusInterfaceBaseAsync, Type[DbusInterfaceBaseAsync]],
+        interface: InterfacesInputElements,
         properties_changed_data: DBUS_PROPERTIES_CHANGED_TYPING,
-        on_unknown_member: Literal['error', 'ignore', 'reuse'] = 'error',
+        on_unknown_member: OnUnknownMember = 'error',
 ) -> Dict[str, Any]:
     interface_name, changed_properties, invalidated_properties = (
         properties_changed_data
@@ -75,17 +91,16 @@ def parse_properties_changed(
 
 
 def _create_interfaces_map(
-    interfaces_iter: Iterable[
-        Union[
-            DbusInterfaceBaseAsync,
-            Type[DbusInterfaceBaseAsync],
-        ]
-    ]
-) -> Dict[FrozenSet[str], Type[DbusInterfaceBaseAsync]]:
-    interfaces_to_class_map: Dict[
-        FrozenSet[str],
-        Type[DbusInterfaceBaseAsync],
-    ] = {}
+    interfaces: InterfacesInput,
+) -> InterfacesToClassMap:
+
+    if isinstance(interfaces,
+                  (DbusInterfaceBaseAsync, type)):
+        interfaces_iter = iter((interfaces, ))
+    else:
+        interfaces_iter = iter(interfaces)
+
+    interfaces_to_class_map: InterfacesToClassMap = {}
 
     for interface in interfaces_iter:
         interface_names_set = frozenset(
@@ -101,51 +116,74 @@ def _create_interfaces_map(
     return interfaces_to_class_map
 
 
+def _get_class_from_interfaces(
+    interfaces_to_class_map: InterfacesToClassMap,
+    interface_names_iter: Iterable[str],
+    raise_key_error: bool,
+) -> Optional[Type[DbusInterfaceBaseAsync]]:
+    class_set = frozenset(interface_names_iter) - SKIP_INTERFACES
+    try:
+        return interfaces_to_class_map[class_set]
+    except KeyError:
+        if raise_key_error:
+            raise
+
+        return None
+
+
+def _get_member_map_from_class(
+    python_class: Optional[Type[DbusInterfaceBaseAsync]],
+) -> Dict[str, Dict[str, str]]:
+    if python_class is None:
+        return {}
+    else:
+        return {
+            interface_name: meta.dbus_member_to_python_attr
+            for interface_name, meta in
+            python_class._dbus_iter_interfaces_meta()
+        }
+
+
+def _translate_and_merge_members(
+    properties_data: Dict[str, Dict[str, Any]],
+    dbus_to_python_map: Dict[str, Dict[str, str]],
+    on_unknown_member: OnUnknownMember,
+) -> Dict[str, Any]:
+    python_properties: Dict[str, Any] = {}
+    for interface_name, properties in properties_data.items():
+        interface_member_map = dbus_to_python_map.get(
+            interface_name, {},
+        )
+        python_properties.update(
+            _parse_properties_vardict(
+                interface_member_map,
+                properties,
+                on_unknown_member,
+            )
+        )
+
+    return python_properties
+
+
 def parse_interfaces_added(
-    interfaces: Union[
-        Union[
-            DbusInterfaceBaseAsync,
-            Type[DbusInterfaceBaseAsync],
-        ],
-        Iterable[
-            Union[
-                DbusInterfaceBaseAsync,
-                Type[DbusInterfaceBaseAsync],
-            ],
-        ],
-    ],
+    interfaces: InterfacesInput,
     interfaces_added_data: Tuple[str, Dict[str, Dict[str, Any]]],
-    on_unknown_interface: Literal['error', 'none'] = 'error',
-    on_unknown_member: Literal['error', 'ignore', 'reuse'] = 'error',
+    on_unknown_interface: OnUnknownInterface = 'error',
+    on_unknown_member: OnUnknownMember = 'error',
 ) -> Tuple[str, Optional[Type[DbusInterfaceBaseAsync]], Dict[str, Any]]:
 
-    if isinstance(interfaces,
-                  (DbusInterfaceBaseAsync, type)):
-        interfaces_iter = iter((interfaces, ))
-    else:
-        interfaces_iter = iter(interfaces)
-
-    interfaces_to_class_map = _create_interfaces_map(interfaces_iter)
+    interfaces_to_class_map = _create_interfaces_map(interfaces)
 
     path, properties_data = interfaces_added_data
 
-    class_set = frozenset(properties_data.keys()) - SKIP_INTERFACES
-    try:
-        python_class = interfaces_to_class_map[class_set]
-        dbus_to_python_member_map: Dict[str, Dict[str, str]] = (
-            {
-                interface_name: meta.dbus_member_to_python_attr
-                for interface_name, meta in
-                python_class._dbus_iter_interfaces_meta()
-            }
+    python_class = (
+        _get_class_from_interfaces(
+            interfaces_to_class_map,
+            properties_data.keys(),
+            on_unknown_interface == "error",
         )
-    except KeyError:
-        if on_unknown_interface == 'error':
-            raise
-
-        python_class = None
-        dbus_to_python_member_map = {}
-
+    )
+    dbus_to_python_member_map = _get_member_map_from_class(python_class)
     python_properties: Dict[str, Any] = {}
     for interface_name, properties in properties_data.items():
         interface_member_map = dbus_to_python_member_map.get(
@@ -159,49 +197,74 @@ def parse_interfaces_added(
             )
         )
 
-    return path, python_class, python_properties
+    return (
+        path,
+        python_class,
+        _translate_and_merge_members(
+            properties_data,
+            dbus_to_python_member_map,
+            on_unknown_member,
+        ),
+    )
 
 
 def parse_interfaces_removed(
-    interfaces: Union[
-        Union[
-            DbusInterfaceBaseAsync,
-            Type[DbusInterfaceBaseAsync],
-        ],
-        Iterable[
-            Union[
-                DbusInterfaceBaseAsync,
-                Type[DbusInterfaceBaseAsync],
-            ],
-        ],
-    ],
+    interfaces: InterfacesInput,
     interfaces_removed_data: Tuple[str, List[str]],
-    on_unknown_interface: Literal['error', 'none'] = 'error',
+    on_unknown_interface: OnUnknownInterface = 'error',
 ) -> Tuple[str, Optional[Type[DbusInterfaceBaseAsync]]]:
-    if isinstance(interfaces,
-                  (DbusInterfaceBaseAsync, type)):
-        interfaces_iter = iter((interfaces, ))
-    else:
-        interfaces_iter = iter(interfaces)
 
-    interfaces_to_class_map = _create_interfaces_map(interfaces_iter)
+    interfaces_to_class_map = _create_interfaces_map(interfaces)
 
     path, interfaces_removed = interfaces_removed_data
 
-    class_set = frozenset(interfaces_removed) - SKIP_INTERFACES
-    try:
-        python_class = interfaces_to_class_map[class_set]
-    except KeyError:
-        if on_unknown_interface == 'error':
-            raise
-
-        python_class = None
+    python_class = (
+        _get_class_from_interfaces(
+            interfaces_to_class_map,
+            interfaces_removed,
+            on_unknown_interface == "error",
+        )
+    )
 
     return path, python_class
 
 
+def parse_get_managed_objects(
+    interfaces: InterfacesInput,
+    managed_objects_data: Dict[str, Dict[str, Dict[str, Any]]],
+    on_unknown_interface: OnUnknownInterface = 'error',
+    on_unknown_member: OnUnknownMember = 'error',
+) -> ParseGetManaged:
+
+    interfaces_to_class_map = _create_interfaces_map(interfaces)
+
+    managed_objects_map: ParseGetManaged = {}
+
+    for path, properties_data in managed_objects_data.items():
+        python_class = (
+            _get_class_from_interfaces(
+                interfaces_to_class_map,
+                properties_data.keys(),
+                on_unknown_interface == "error",
+            )
+        )
+        dbus_to_python_member_map = _get_member_map_from_class(python_class)
+
+        managed_objects_map[path] = (
+            python_class,
+            _translate_and_merge_members(
+                properties_data,
+                dbus_to_python_member_map,
+                on_unknown_member,
+            ),
+        )
+
+    return managed_objects_map
+
+
 __all__ = (
     'parse_properties_changed',
     'parse_interfaces_added',
     'parse_interfaces_removed',
+    'parse_get_managed_objects',
 )
diff --git a/test/test_object_manager.py b/test/test_object_manager.py
