add restart_timer argument to GenericSignalThrottler.flush (#206)

src/superqt/utils/_throttler.py

@@ -139,18 +139,28 @@ class GenericSignalThrottler(QObject):
         """Cancel any pending emissions."""
         self._hasPendingEmission = False
 
-    def flush(self) -> None:
-        """Force emission of any pending emissions."""
-        self._maybeEmitTriggered()
+    def flush(self, restart_timer: bool = True) -> None:
+        """
+        Force emission of any pending emissions.
+
+        Parameters
+        ----------
+        restart_timer : bool
+            Whether to restart the timer after flushing.
+            Defaults to True.
+        """
+        self._maybeEmitTriggered(restart_timer=restart_timer)
 
     def _emitTriggered(self) -> None:
         self._hasPendingEmission = False
         self.triggered.emit()
         self._timer.start()
 
-    def _maybeEmitTriggered(self) -> None:
+    def _maybeEmitTriggered(self, restart_timer=True) -> None:
         if self._hasPendingEmission:
             self._emitTriggered()
+        if not restart_timer:
+            self._timer.stop()
 
     Kind = Kind
     EmissionPolicy = EmissionPolicy

tests/test_throttler.py

@@ -27,6 +27,41 @@ def test_debounced(qtbot):
     assert mock2.call_count == 10
 
 
+@pytest.mark.usefixtures("qapp")
+def test_stop_timer_simple():
+    mock = Mock()
+
+    @qdebounced(timeout=5)
+    def f1() -> str:
+        mock()
+
+    f1()
+    assert f1._timer.isActive()
+    mock.assert_not_called()
+    f1.flush(restart_timer=False)
+    assert not f1._timer.isActive()
+    mock.assert_called_once()
+
+
+@pytest.mark.usefixtures("qapp")
+def test_stop_timer_no_event_pending():
+    mock = Mock()
+
+    @qdebounced(timeout=5)
+    def f1() -> str:
+        mock()
+
+    f1()
+    assert f1._timer.isActive()
+    mock.assert_not_called()
+    f1.flush()
+    assert f1._timer.isActive()
+    mock.assert_called_once()
+    f1.flush(restart_timer=False)
+    assert not f1._timer.isActive()
+    mock.assert_called_once()
+
+
 def test_debouncer_method(qtbot):
     class A(QObject):
         def __init__(self):