pyqt_reactive.core.background_task
Unified background task with cancellation, debounce, and cleanup.
Classes
|
Unified background task with cancellation, debounce, and cleanup. |
Manages background task lifecycle for a widget. |
- class pyqt_reactive.core.background_task.BackgroundTask(target: Callable[[...], Any], args: Tuple = (), kwargs: dict = None, debounce_ms: int = 0, parent=None)[source]
Unified background task with cancellation, debounce, and cleanup.
- Usage:
task = BackgroundTask(target=my_func, args=(a, b)) task.result_ready.connect(on_success) task.error_occurred.connect(on_error) # Receives Exception, not str task.start()
# Later: task.cancel() # Safe cancellation
- Error handling:
- def on_error(e: Exception):
logger.exception(“Failed”, exc_info=e) # Full traceback show_message(str(e)) # User-friendly string if isinstance(e, TimeoutError): … # Type checking
- result_ready
int = …, arguments: Sequence = …) -> PYQT_SIGNAL
types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.
- Type:
pyqtSignal(*types, name
- Type:
str = …, revision
- error_occurred
int = …, arguments: Sequence = …) -> PYQT_SIGNAL
types is normally a sequence of individual types. Each type is either a type object or a string that is the name of a C++ type. Alternatively each type could itself be a sequence of types each describing a different overloaded signal. name is the optional C++ name of the signal. If it is not specified then the name of the class attribute that is bound to the signal is used. revision is the optional revision of the signal that is exported to QML. If it is not specified then 0 is used. arguments is the optional sequence of the names of the signal’s arguments.
- Type:
pyqtSignal(*types, name
- Type:
str = …, revision
- class pyqt_reactive.core.background_task.BackgroundTaskManager[source]
Manages background task lifecycle for a widget.
Handles: - Cancelling previous task before starting new one - Cleanup on widget close - Debounce across rapid calls - Button state management (disable during operation, auto-restore)
- Usage in widget:
self._task_manager = BackgroundTaskManager()
- def refresh_data(self):
- self._task_manager.run(
target=self.service.fetch_data, args=(self.query,), button=self.refresh_button, button_loading_text=”Loading…”, on_success=self._on_data_ready, on_error=self._on_error, debounce_ms=200
)
- def closeEvent(self, event):
self._task_manager.cleanup() super().closeEvent(event)
- run(target: Callable[[...], Any], args: Tuple = (), kwargs: dict = None, on_success: Callable[[Any], None] = None, on_error: Callable[[Exception], None] = None, debounce_ms: int = 0, button: QPushButton = None, button_loading_text: str = None) BackgroundTask | None[source]
Run a background task, cancelling any previous one.
- Parameters:
target – Function to execute in background
args – Positional arguments for target
kwargs – Keyword arguments for target
on_success – Callback for successful result
on_error – Callback for error (receives Exception, not str)
debounce_ms – Minimum time between runs (skip if too soon)
button – Button to disable during operation (auto-restored on complete)
button_loading_text – Text while loading (default: original + “…”)
- Returns:
BackgroundTask if started, None if debounced out