Add sdbus.utils.inspect.inspect_dbus_attached_bus function

igo95862 · igo95862 · commit f572c92a3353 · 2025-09-13T22:45:02.000+01:00
Returns the D-Bus bus used by the object.
diff --git a/src/sdbus/utils/inspect.py b/src/sdbus/utils/inspect.py
@@ -110,6 +110,32 @@ def inspect_dbus_path(
         raise TypeError(f"Expected D-Bus object got {obj!r}")
 
 
+def inspect_dbus_bus(
+    obj: Union[DbusInterfaceBase, DbusInterfaceBaseAsync]
+) -> Optional[SdBus]:
+    """Return D-Bus bus used by the object.
+
+    If called on D-Bus proxies or exported local D-Bus objects returns
+    bus object.
+
+    If called on local D-Bus objects that had not been exported returns None.
+
+    If called on an object that is unrelated to D-Bus raises ``TypeError``.
+
+    :param obj:
+        Object to inspect.
+    :returns:
+        D-Bus bus object.
+
+    *New in version 0.14.1.*
+    """
+    if isinstance(obj, (DbusInterfaceBase, DbusInterfaceBaseAsync)):
+        return obj._dbus.attached_bus
+    else:
+        raise TypeError(f"Expected D-Bus object got {obj!r}")
+
+
 __all__ = (
+    'inspect_dbus_bus',
     "inspect_dbus_path",
 )
diff --git a/test/test_sdbus_utils.py b/test/test_sdbus_utils.py
@@ -22,7 +22,7 @@
 from unittest import TestCase
 
 from sdbus.unittest import IsolatedDbusTestCase
-from sdbus.utils.inspect import inspect_dbus_path
+from sdbus.utils.inspect import inspect_dbus_bus, inspect_dbus_path
 from sdbus.utils.parse import parse_get_managed_objects
 
 from sdbus import (
@@ -258,3 +258,24 @@ def test_inspect_dbus_path_async_local(self) -> None:
 
         with self.assertRaisesRegex(LookupError, "is not attached to bus"):
             inspect_dbus_path(local_obj, new_bus)
+
+    def test_inspect_attached_bus(self) -> None:
+        proxy = DbusInterfaceCommon("example.org", TEST_PATH)
+
+        self.assertIs(inspect_dbus_bus(proxy), self.bus)
+
+        with self.assertRaises(TypeError):
+            inspect_dbus_bus(object())  # type: ignore[arg-type]
+
+    def test_inspect_attached_bus_async(self) -> None:
+        proxy = DbusInterfaceCommonAsync.new_proxy("example.org", TEST_PATH)
+
+        self.assertIs(inspect_dbus_bus(proxy), self.bus)
+
+        local_obj = FooBarAsync()
+
+        self.assertIsNone(inspect_dbus_bus(local_obj))
+
+        local_obj.export_to_dbus("/")
+
+        self.assertIs(inspect_dbus_bus(local_obj), self.bus)
