Add logic to allow napari to know that plugin actions are being done

src/napari_plugin_manager/_tests/test_qt_plugin_dialog.py

@@ -672,3 +672,26 @@ def test_import_plugins(plugin_dialog, tmp_path, qtbot):
     path.write_text('requests\npackaging\n')
     with qtbot.waitSignal(plugin_dialog.installer.allFinished, timeout=60_000):
         plugin_dialog.import_plugins(str(path))
+
+
+def test_query_status(plugin_dialog, monkeypatch):
+    status, description = plugin_dialog.query_status()
+    assert status == qt_plugin_dialog.Status.DONE
+    assert not description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        '_queue',
+        ['mock'],
+    )
+    status, description = plugin_dialog.query_status()
+    assert status == qt_plugin_dialog.Status.BUSY
+    assert description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        '_queue',
+        ['mock', 'other-mock'],
+    )
+    assert status == qt_plugin_dialog.Status.BUSY
+    assert description

src/napari_plugin_manager/base_qt_package_installer.py

@@ -478,6 +478,7 @@ def cancel_all(self):
                 process.errorOccurred.disconnect(self._on_error_occurred)
 
             self._end_process(process)
+            process.waitForFinished()
 
         self._queue.clear()
         self._current_process = None

src/napari_plugin_manager/base_qt_plugin_dialog.py

@@ -1,8 +1,10 @@
 import contextlib
 import importlib.metadata
 import os
+import uuid
 import webbrowser
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
+from enum import Enum, auto
 from functools import partial
 from logging import getLogger
 from typing import (
@@ -61,6 +63,14 @@
 log = getLogger(__name__)
 
 
+class Status(Enum):
+    PENDING = auto()
+    BUSY = auto()
+    DONE = auto()
+    CANCELLED = auto()
+    FAILED = auto()
+
+
 class PackageMetadataProtocol(Protocol):
     """
     Protocol class defining the minimum atributtes/properties needed for package metadata.
@@ -1048,6 +1058,7 @@ def __init__(self, parent=None, prefix=None) -> None:
         self._plugin_data = []  # Store all plugin data
         self._filter_texts = []
         self._filter_idxs_cache = set()
+        self._task_status_id = None
         self.worker = None
         self._plugin_data_map = {}
         self._add_items_timer = QTimer(self)
@@ -1130,13 +1141,33 @@ def _update_theme(self, event: Any) -> None:
         """
         raise NotImplementedError
 
+    def _register_task_status(self):
+        status, description = self.query_status()
+
+        if self._task_status_id is not None:
+            self._update_task_status(status, description=description)
+            return
+
+        self._task_status_id = self.register_task_status(
+            status, description, cancel_callback=self.installer.cancel_all
+        )
+
+    def _update_task_status(
+        self, status: Status = Status.DONE, description: str = ''
+    ):
+        if self._task_status_id is not None:
+            self.update_task_status(
+                self._task_status_id, status, description=description
+            )
+
     def _on_installer_start(self):
         """Updates dialog buttons and status when installing a plugin."""
         self.cancel_all_btn.setVisible(True)
         self.working_indicator.show()
         self.process_success_indicator.hide()
         self.process_error_indicator.hide()
         self.refresh_button.setDisabled(True)
+        self._register_task_status()
 
     def _on_process_finished(self, process_finished_data: ProcessFinishedData):
         action = process_finished_data['action']
@@ -1210,6 +1241,7 @@ def _on_installer_all_finished(self, exit_codes):
                     self._trans('Plugin Manager: process completed\n')
                 )
 
+        self._update_task_status()
         self.search()
 
     def _add_to_installed(
@@ -1774,7 +1806,7 @@ def search(self, text: str | None = None, skip=False) -> None:
         if len(text.strip()) == 0:
             self.installed_list.filter('')
             self.available_list.hideAll()
-            self._plugin_queue = None
+            self._plugin_queue = []
             self._add_items_timer.stop()
             self._plugins_found = 0
         else:
@@ -1792,7 +1824,7 @@ def search(self, text: str | None = None, skip=False) -> None:
                 self._plugins_found = len(items)
                 self._add_items_timer.start()
             else:
-                self._plugin_queue = None
+                self._plugin_queue = []
                 self._add_items_timer.stop()
                 self._plugins_found = 0
 
@@ -1811,6 +1843,7 @@ def refresh(self, clear_cache: bool = False):
         self._plugin_queue = []
         self._plugin_data = []
         self._plugin_data_map = {}
+        self._latest_status = None
 
         self.installed_list.clear()
         self.available_list.clear()
@@ -1868,4 +1901,30 @@ def import_plugins(self, fpath: str) -> None:
         plugins = [p for p in plugins if p]
         self._install_packages(plugins)
 
+    def register_task_status(
+        self,
+        status: Status,
+        description: str,
+        cancel_callback: Callable | None = None,
+    ) -> uuid.UUID:
+        """Register a task status for the plugin manager."""
+        raise NotImplementedError
+
+    def update_task_status(
+        self, task_status_id: uuid.UUID, status: Status, description: str = ''
+    ) -> bool:
+        """Update task status for the plugin manager."""
+        raise NotImplementedError
+
+    def query_status(self) -> tuple[Status, str]:
+        """
+        Return the current status of plugins installations.
+
+        Returns
+        -------
+        A tuple containing the current status (`Status`) and a description.
+
+        """
+        raise NotImplementedError
+
     # endregion - Public methods

src/napari_plugin_manager/qt_plugin_dialog.py

@@ -1,4 +1,6 @@
 import sys
+import uuid
+from collections.abc import Callable
 from pathlib import Path
 
 import napari.plugins
@@ -37,6 +39,31 @@
 from napari_plugin_manager.qt_package_installer import NapariInstallerQueue
 from napari_plugin_manager.utils import is_conda_package
 
+try:
+    from napari.utils.task_status import (
+        Status,
+        register_task_status,
+        update_task_status,
+    )
+except ImportError:
+    from napari_plugin_manager.base_qt_plugin_dialog import Status
+
+    def register_task_status(
+        provider: str,
+        task_status: Status,
+        description: str,
+        cancel_callback: Callable | None = None,
+    ) -> uuid.UUID:
+        pass
+
+    def update_task_status(
+        task_status_id: uuid.UUID,
+        status: Status,
+        description: str = '',
+    ) -> bool:
+        pass
+
+
 # Scaling factor for each list widget item when expanding.
 STYLES_PATH = Path(__file__).parent / 'styles.qss'
 DISMISS_WARN_PYPI_INSTALL_DLG = False
@@ -265,6 +292,43 @@ def _show_warning(self, warning):
     def _trans(self, text, **kwargs):
         return trans._(text, **kwargs)
 
+    def register_task_status(
+        self,
+        task_status: Status,
+        description: str,
+        cancel_callback: Callable | None = None,
+    ) -> uuid.UUID:
+        return register_task_status(
+            'napari-plugin-manager',
+            task_status,
+            description,
+            cancel_callback=cancel_callback,
+        )
+
+    def update_task_status(
+        self,
+        task_status_id: uuid.UUID,
+        status: Status,
+        description: str = '',
+    ) -> bool:
+        return update_task_status(
+            task_status_id, status, description=description
+        )
+
+    def query_status(self) -> tuple[Status, str]:
+        if self.installer.hasJobs():
+            task_status = Status.BUSY
+            description = trans._n(
+                'The plugin manager is currently busy with {n} task.',
+                'The plugin manager is currently busy with {n} tasks.',
+                n=self.installer.currentJobs(),
+            )
+        else:
+            task_status = Status.DONE
+            description = ''
+
+        return task_status, description
+
 
 if __name__ == '__main__':
     from qtpy.QtWidgets import QApplication