Added sdbus.utils.parse_interfaces_removed function

Parses object manager interfaces_removed signal and returns
the path of removed object and python class.

docs/asyncio_api.rst

@@ -188,6 +188,9 @@ Classes
         Signal when existing object or and interface of
         existing object is removed.
 
+        :py:func:`sdbus.utils.parse_interfaces_removed` can be used
+        to make signal data easier to work with.
+
         Signal data is:
 
         Object path : str

docs/utils.rst

@@ -42,3 +42,20 @@ Parsing utilities
     :rtype: Tuple[str, Optional[Type[DbusInterfaceBaseAsync]], Dict[str, Any]]
     :returns: Path of new added object, object's class (or ``None``) and dictionary
             of python translated members and their values.
+
+.. py:function:: parse_interfaces_removed(interfaces, interfaces_removed_data, on_unknown_interface='error')
+
+    Parse data from :py:meth:`interfaces_added <sdbus.DbusObjectManagerInterfaceAsync.interfaces_removed>` signal.
+
+    Takes an iterable of D-Bus interface classes (or a single class) and the signal data.
+    Returns the path of removed object andthe class of the added object.
+    (if it matched one of passed interface classes)
+
+    :param Iterable[DbusInterfaceBaseAsync] interfaces: Possible interfaces that were removed.
+        Can accept classes with multiple interfaces defined.
+    :param Tuple interfaces_added_data: Tuple caught from signal.
+    :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: Tuple[str, Optional[Type[DbusInterfaceBaseAsync]]]
+    :returns: Path of removed object and object's class (or ``None``).

src/sdbus/utils.py

@@ -24,6 +24,7 @@
     Dict,
     FrozenSet,
     Iterable,
+    List,
     Literal,
     Optional,
     Tuple,
@@ -61,34 +62,19 @@ def parse_properties_changed(
 ))
 
 
-def parse_interfaces_added(
-    interfaces: Union[
+def _create_interfaces_map(
+    interfaces_iter: Iterable[
         Union[
             DbusInterfaceBaseAsync,
             Type[DbusInterfaceBaseAsync],
-        ],
-        Iterable[
-            Union[
-                DbusInterfaceBaseAsync,
-                Type[DbusInterfaceBaseAsync],
-            ],
-        ],
-    ],
-    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',
-) -> Tuple[str, Optional[Type[DbusInterfaceBaseAsync]], Dict[str, Any]]:
+        ]
+    ]
+) -> Dict[FrozenSet[str], Type[DbusInterfaceBaseAsync]]:
     interfaces_to_class_map: Dict[
         FrozenSet[str],
         Type[DbusInterfaceBaseAsync],
     ] = {}
 
-    if isinstance(interfaces,
-                  (DbusInterfaceBaseAsync, type)):
-        interfaces_iter = iter((interfaces, ))
-    else:
-        interfaces_iter = iter(interfaces)
-
     for interface in interfaces_iter:
         if (
             isinstance(interface, DbusInterfaceBaseAsync)
@@ -107,6 +93,35 @@ def parse_interfaces_added(
         else:
             raise TypeError('Expected D-Bus interface, got: ', interface)
 
+    return interfaces_to_class_map
+
+
+def parse_interfaces_added(
+    interfaces: Union[
+        Union[
+            DbusInterfaceBaseAsync,
+            Type[DbusInterfaceBaseAsync],
+        ],
+        Iterable[
+            Union[
+                DbusInterfaceBaseAsync,
+                Type[DbusInterfaceBaseAsync],
+            ],
+        ],
+    ],
+    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',
+) -> 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)
+
     path, properties_data = interfaces_added_data
 
     class_set = frozenset(properties_data.keys()) - SKIP_INTERFACES
@@ -133,6 +148,44 @@ def parse_interfaces_added(
     return path, python_class, python_properties
 
 
+def parse_interfaces_removed(
+    interfaces: Union[
+        Union[
+            DbusInterfaceBaseAsync,
+            Type[DbusInterfaceBaseAsync],
+        ],
+        Iterable[
+            Union[
+                DbusInterfaceBaseAsync,
+                Type[DbusInterfaceBaseAsync],
+            ],
+        ],
+    ],
+    interfaces_removed_data: Tuple[str, List[str]],
+    on_unknown_interface: Literal['error', 'none'] = '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)
+
+    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
+
+    return path, python_class
+
+
 __all__ = (
     'parse_properties_changed',
     'parse_interfaces_added',

test/test_object_manager.py

@@ -24,7 +24,7 @@
 from typing import Any, Dict, List, Tuple
 
 from sdbus.unittest import IsolatedDbusTestCase
-from sdbus.utils import parse_interfaces_added
+from sdbus.utils import parse_interfaces_added, parse_interfaces_removed
 
 from sdbus import (
     DbusInterfaceCommonAsync,
@@ -143,7 +143,7 @@ def test_expot_with_no_manager(self) -> None:
             managed_object,
         )
 
-    async def test_parse_interfaces_added(self) -> None:
+    async def test_parse_interfaces_added_removed(self) -> None:
         MANAGED_TWO_INTERFACE_NAME = MANAGED_INTERFACE_NAME + 'Two'
 
         class ManagedTwoInterface(
@@ -174,6 +174,14 @@ async def catch_interfaces_added() -> Tuple[str,
 
         catch_added_task = loop.create_task(catch_interfaces_added())
 
+        async def catch_interfaces_removed() -> Tuple[str, List[str]]:
+            async for x in object_manager_connection.interfaces_removed:
+                return x
+
+            raise RuntimeError
+
+        catch_removed_task = loop.create_task(catch_interfaces_removed())
+
         await sleep(0)
 
         managed_object = ManagedTwoInterface()
@@ -182,7 +190,7 @@ async def catch_interfaces_added() -> Tuple[str,
 
         caught_added = await wait_for(catch_added_task, timeout=0.5)
 
-        with self.subTest('Parse class'):
+        with self.subTest('Parse added class'):
             path, python_class, python_properties = (
                 parse_interfaces_added(ManagedTwoInterface, caught_added)
             )
@@ -192,7 +200,7 @@ async def catch_interfaces_added() -> Tuple[str,
             self.assertIn('test_str', python_properties)
             self.assertIn('test_int', python_properties)
 
-        with self.subTest('Parse object'):
+        with self.subTest('Parse added object'):
             path, python_class, python_properties = (
                 parse_interfaces_added(managed_object, caught_added)
             )
@@ -202,7 +210,7 @@ async def catch_interfaces_added() -> Tuple[str,
             self.assertIn('test_str', python_properties)
             self.assertIn('test_int', python_properties)
 
-        with self.subTest('Parse iterable'):
+        with self.subTest('Parse added iterable'):
             path, python_class, python_properties = (
                 parse_interfaces_added(
                     (ManagedInterface, ManagedTwoInterface),
@@ -214,7 +222,7 @@ async def catch_interfaces_added() -> Tuple[str,
             self.assertIn('test_str', python_properties)
             self.assertIn('test_int', python_properties)
 
-        with self.subTest('Parse unknown'):
+        with self.subTest('Parse added unknown'):
             with self.assertRaises(KeyError):
                 path, python_class, python_properties = (
                     parse_interfaces_added(
@@ -244,3 +252,39 @@ async def catch_interfaces_added() -> Tuple[str,
             self.assertIsNone(python_class)
             self.assertIn('TestStr', python_properties)
             self.assertIn('TestInt', python_properties)
+
+        object_manager.remove_managed_object(managed_object)
+
+        interfaces_removed_data = await wait_for(
+            catch_removed_task, timeout=1)
+
+        with self.subTest('Parse removed class'):
+            path, python_class = (
+                parse_interfaces_removed(
+                    ManagedTwoInterface,
+                    interfaces_removed_data,
+                )
+            )
+
+            self.assertEqual(path, MANAGED_PATH)
+            self.assertEqual(python_class, ManagedTwoInterface)
+
+        with self.subTest('Parse removed unknown'):
+            with self.assertRaises(KeyError):
+                path, python_class = (
+                    parse_interfaces_removed(
+                        ManagedInterface,
+                        interfaces_removed_data,
+                    )
+                )
+
+            path, python_class = (
+                parse_interfaces_removed(
+                    ManagedInterface,
+                    interfaces_removed_data,
+                    on_unknown_interface='none',
+                )
+            )
+
+            self.assertEqual(path, MANAGED_PATH)
+            self.assertIsNone(python_class)