feat: add the ability for signal callbacks to ignore arguments

Author: its-darsh

fabric/core/service.py

@@ -27,6 +27,7 @@
     snake_case_to_kebab_case,
     kebab_case_to_snake_case,
     get_function_annotations,
+    make_arguments_ignorable,
 )
 
 OldSignal = gi._signalhelper.Signal
@@ -252,9 +253,18 @@ def emit(self, /, *args: P.args, **kwargs: P.kwargs) -> Optional[R]:
 
     # TODO: make it hint self's instance for the callback
     def connect(
-        self, callback: Callable[Concatenate[GObject.Object, P], Any], *args, **kwargs
+        self,
+        callback: Callable[Concatenate[G, P], Any] | Callable,
+        *args,
+        ignore_missing: bool = True,
     ) -> int:
-        return self.instance.connect(self.name, callback)  # type: ignore
+        return Service.connect(
+            self.instance,  # type: ignore
+            self.name,
+            callback,
+            *args,
+            ignore_missing=ignore_missing,
+        )
 
 
 class Signal(Generic[G, P, R]):
@@ -372,6 +382,7 @@ def installer(klass):
 
         # all aboard...
         setattr(klass, "__gsignals__", klass_signals)
+        return
 
 
 @dataclass
@@ -503,11 +514,15 @@ def emit(self, signal_name: str, *args, **kwargs) -> None:
     def connect(
         self,
         signal_name: str,
-        callback: Callable[Concatenate[Self, P], Any],
+        callback: Callable[Concatenate[Self, P], Any] | Callable,
         *args,
-        **kwargs,
+        ignore_missing: bool = True,
     ) -> int:
-        return super().connect(signal_name, callback, *args, **kwargs)  # type: ignore
+        return super().connect(
+            signal_name,
+            make_arguments_ignorable(callback) if ignore_missing else callback,
+            *args,
+        )  # type: ignore
 
     @staticmethod
     def filter_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:

fabric/network/service.py

@@ -0,0 +1,62 @@
+import gi
+from loguru import logger
+from collections.abc import Callable
+from typing import ParamSpec, Literal, Concatenate, Any
+from fabric.core.service import Service, Signal, Property
+from fabric.utils.helpers import (
+    bulk_connect,
+    get_enum_member_name,
+    snake_case_to_kebab_case,
+    bridge_signal,
+)
+
+from gi.repository import Gio
+
+try:
+    gi.require_version("GnomeBluetooth", "3.0")
+    from gi.repository import NM as NetworkManager
+except Exception:
+    raise ImportError("gnome-bluetooth-3 is not installed, please install it first")
+
+
+class Connection(Service):  # base block for both wlan & eth
+    @Property(NetworkManager.ActiveConnection, flags="readable")
+    def connection(self) -> NetworkManager.ActiveConnection:
+        return self._connection
+
+    @Property(str, flags="readable")
+    def connection_type(self) -> Literal["wifi", "wired", "unknown"]:
+        return self._connection_type  # type: ignore
+
+    @Property(str, flags="readable")
+    def status(
+        self,
+    ) -> Literal["connected", "connecting", "disconnected", "disconnecting", "unknown"]:
+        return get_enum_member_name(
+            self._connection.get_state(),
+            {
+                "ACTIVATING": "connecting",
+                "ACTIVATED": "connected",
+                "DEACTIVATING": "disconnecting",
+                "DEACTIVATED": "disconnected",
+            },
+            default="unknown",
+        ).lower()  # type: ignore
+
+    def __init__(
+        self,
+        connection: NetworkManager.ActiveConnection,
+        connection_type: str,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+
+        self._connection = connection
+        self._connection_type = connection_type
+
+        self._status = "unknown"
+
+
+class Network(Service):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)

fabric/utils/helpers.py

@@ -589,7 +589,9 @@ def monitor_file(
 
 
 def cooldown(
-    cooldown_time:  int | float , error: Callable | None = None, return_error: bool = False
+    cooldown_time: int | float,
+    error: Callable | None = None,
+    return_error: bool = False,
 ):
     """
     Decorator function that adds a cooldown period to a given function
@@ -856,6 +858,25 @@ def get_function_annotations(
     return FunctionAnnotations(args, return_type)
 
 
+def make_arguments_ignorable(func: Callable[..., T]) -> Callable[..., T]:
+    params = inspect.signature(func).parameters.values()
+    if any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in params):
+        return func  # no
+
+    args_len = sum(
+        1
+        for p in params
+        if p.kind
+        in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
+    )
+
+    @wraps(func)
+    def wrapper(*passed_args):
+        return func(*passed_args[:args_len])
+
+    return wrapper
+
+
 def truncate(string: str, max_length: int, suffix: str = "...") -> str:
     return (
         string

tests/service.3.py

@@ -196,6 +196,116 @@ def onSignal(signalName):
         self.assertEqual(emittedSignals, expectedOutput)
         emittedSignals = []
 
+    def testServiceSignalsConnectAndEmit5(self):
+        # ignorable arguments
+        emittedSignals: list[str] = []
+        expectedOutput = [
+            "int-signal",
+            "str-signal",
+            "bool-signal",
+            "boxed-signal",
+            "complex-signal",
+        ]
+
+        def onSignal(signalName):
+            emittedSignals.append(signalName)
+
+        service = self.ServiceWithSignals(
+            on_int_signal=lambda: onSignal("int-signal"),
+            on_str_signal=lambda: onSignal("str-signal"),
+            on_bool_signal=lambda: onSignal("bool-signal"),
+            on_boxed_signal=lambda: onSignal("boxed-signal"),
+            on_complex_signal=lambda: onSignal("complex-signal"),
+        )
+
+        service.emit("int-signal", 42)
+        service.emit("str-signal", "42")
+        service.emit("bool-signal", True)
+        service.emit("boxed-signal", object)
+        service.emit("complex-signal", 42, "42", True, object)
+
+        self.assertEqual(emittedSignals, expectedOutput)
+        emittedSignals = []
+
+    def testServiceSignalsConnectAndEmit6(self):
+        # ignorable arguments
+        emittedSignals: list[str] = []
+        expectedOutput = [
+            "int-signal",
+            "str-signal",
+            "bool-signal",
+            "boxed-signal",
+            "complex-signal",
+        ]
+
+        def onSignal(signalName):
+            emittedSignals.append(signalName)
+
+        service = self.ServiceWithSignals(
+            on_int_signal=lambda s: self.assertEqual(s, service)
+            or onSignal("int-signal"),
+            on_str_signal=lambda s: self.assertEqual(s, service)
+            or onSignal("str-signal"),
+            on_bool_signal=lambda s: self.assertEqual(s, service)
+            or onSignal("bool-signal"),
+            on_boxed_signal=lambda s: self.assertEqual(s, service)
+            or onSignal("boxed-signal"),
+            on_complex_signal=lambda s: self.assertEqual(s, service)
+            or onSignal("complex-signal"),
+        )
+
+        service.emit("int-signal", 42)
+        service.emit("str-signal", "42")
+        service.emit("bool-signal", True)
+        service.emit("boxed-signal", object)
+        service.emit("complex-signal", 42, "42", True, object)
+
+        self.assertEqual(emittedSignals, expectedOutput)
+        emittedSignals = []
+
+    def testServiceSignalsConnectAndEmit7(self):
+        # ignorable arguments
+        emittedSignals: list[str] = []
+        expectedOutput = [
+            "int-signal",
+            "str-signal",
+            "bool-signal",
+            "boxed-signal",
+            "complex-signal",
+        ]
+
+        def onSignal(signalName):
+            emittedSignals.append(signalName)
+
+        service = self.ServiceWithSignals(
+            on_int_signal=lambda s, i: self.assertEqual(i, 42)
+            or self.assertEqual(s, service)
+            or onSignal("int-signal"),
+            on_str_signal=lambda s, i: self.assertEqual(i, "42")
+            or self.assertEqual(s, service)
+            or onSignal("str-signal"),
+            on_bool_signal=lambda s, i: self.assertEqual(i, True)
+            or self.assertEqual(s, service)
+            or onSignal("bool-signal"),
+            on_boxed_signal=lambda s, i: self.assertEqual(i, object)
+            or self.assertEqual(s, service)
+            or onSignal("boxed-signal"),
+            on_complex_signal=lambda s, i1, i2, i3, i4: self.assertEqual(
+                (i1, i2, i3, i4), (42, "42", True, object)
+            )
+            or self.assertEqual(s, service)
+            or onSignal("complex-signal"),
+        )
+
+        service.emit("int-signal", 42)
+        service.emit("str-signal", "42")
+        service.emit("bool-signal", True)
+        service.emit("boxed-signal", object)
+        service.emit("complex-signal", 42, "42", True, object)
+
+        self.assertEqual(emittedSignals, expectedOutput)
+        emittedSignals = []
+
 
 if __name__ == "__main__":
     unittest.main()