Add a tasks manager status for plugins actions and napari processes

src/napari/_qt/_qapp_model/qactions/_plugins.py

@@ -32,7 +32,11 @@ def _show_plugin_install_dialog(window: Window) -> None:
     # This callback is only used when this package is available, thus we do not check
     from napari_plugin_manager.qt_plugin_dialog import QtPluginDialog
 
-    QtPluginDialog(window._qt_window).exec_()
+    window._qt_window._plugin_manager_dialog = QtPluginDialog(
+        window._qt_window
+    )
+    if window._qt_window._plugin_manager_dialog is not None:
+        window._qt_window._plugin_manager_dialog.exec_()
 
 
 def _show_plugin_err_reporter(window: Window) -> None:

src/napari/_qt/dialogs/confirm_close_dialog.py

@@ -14,33 +14,44 @@
 
 
 class ConfirmCloseDialog(QDialog):
-    def __init__(self, parent, close_app=False) -> None:
+    def __init__(
+        self,
+        parent,
+        close_app=False,
+        display_checkbox=True,
+        extra_info='',
+    ) -> None:
         super().__init__(parent)
+        extra_info = f'\n\n{extra_info}' if extra_info else ''
         cancel_btn = QPushButton(trans._('Cancel'))
         close_btn = QPushButton(trans._('Close'))
         close_btn.setObjectName('warning_icon_btn')
         icon_label = QWidget()
 
         self.do_not_ask = QCheckBox(trans._('Do not ask in future'))
+        self.do_not_ask.setVisible(display_checkbox)
+        self._display_checkbox = display_checkbox
 
         if close_app:
             self.setWindowTitle(trans._('Close Application?'))
             text = trans._(
-                "Do you want to close the application? ('{shortcut}' to confirm). This will close all Qt Windows in this process",
+                "Do you want to close the application? ('{shortcut}' to confirm). This will close all Qt Windows in this process{extra_info}",
                 shortcut=QKeySequence('Ctrl+Q').toString(
                     QKeySequence.NativeText
                 ),
+                extra_info=extra_info,
             )
             close_btn.setObjectName('error_icon_btn')
             close_btn.setShortcut(QKeySequence('Ctrl+Q'))
             icon_label.setObjectName('error_icon_element')
         else:
             self.setWindowTitle(trans._('Close Window?'))
             text = trans._(
-                "Confirm to close window (or press '{shortcut}')",
+                "Confirm to close window (or press '{shortcut}'){extra_info}",
                 shortcut=QKeySequence('Ctrl+W').toString(
                     QKeySequence.NativeText
                 ),
+                extra_info=extra_info,
             )
             close_btn.setObjectName('warning_icon_btn')
             close_btn.setShortcut(QKeySequence('Ctrl+W'))
@@ -70,6 +81,6 @@ def __init__(self, parent, close_app=False) -> None:
         self.cancel_btn = cancel_btn
 
     def accept(self):
-        if self.do_not_ask.isChecked():
+        if self._display_checkbox and self.do_not_ask.isChecked():
             get_settings().application.confirm_close_window = False
         super().accept()

src/napari/_qt/qt_main_window.py

@@ -92,6 +92,7 @@
     running_as_constructor_app,
 )
 from napari.utils.notifications import Notification
+from napari.utils.task_status import task_status_manager
 from napari.utils.theme import _themes, get_system_theme
 from napari.utils.translations import trans
 
@@ -135,6 +136,7 @@ def __init__(
         super().__init__(parent)
         self._ev = None
         self._window = window
+        self._plugin_manager_dialog = None
         self._qt_viewer = QtViewer(viewer, show_welcome_screen=True)
         self._quit_app = False
 
@@ -490,8 +492,9 @@ def close(self, quit_app=False, confirm_need=False):
             return super().close()
         confirm_need_local = confirm_need and self._is_close_dialog[quit_app]
         self._is_close_dialog[quit_app] = False
+
         # here we save information that we could request confirmation on close
-        # So fi function `close` is called again, we don't ask again but just close
+        # So if function `close` is called again, we don't ask again but just close
         if (
             not confirm_need_local
             or not get_settings().application.confirm_close_window
@@ -602,15 +605,29 @@ def closeEvent(self, event):
 
         Regardless of whether cmd Q, cmd W, or the close button is used...
         """
+        task_status = task_status_manager.get_status()
         if (
+            task_status
+            and ConfirmCloseDialog(
+                self,
+                close_app=False,
+                extra_info='\n'.join(task_status),
+                display_checkbox=False,
+            ).exec_()
+            != QDialog.Accepted
+        ) or (
             event.spontaneous()
             and get_settings().application.confirm_close_window
             and self._qt_viewer.viewer.layers
-            and ConfirmCloseDialog(self, False).exec_() != QDialog.Accepted
+            and ConfirmCloseDialog(self, close_app=False).exec_()
+            != QDialog.Accepted
         ):
             event.ignore()
             return
 
+        if task_status_manager.is_busy():
+            task_status_manager.cancel_all()
+
         self.status_thread.close_terminate()
         self.status_thread.wait()
 
@@ -704,6 +721,7 @@ def __init__(self, viewer: 'Viewer', *, show: bool = True) -> None:
 
         # Connect the Viewer and create the Main Window
         self._qt_window = _QtMainWindow(viewer, self)
+
         qapp.installEventFilter(self._qt_window)
 
         # connect theme events before collecting plugin-provided themes

src/napari/utils/_tests/test_task_status.py

@@ -0,0 +1,46 @@
+from unittest.mock import Mock
+
+from napari.utils.task_status import (
+    Status,
+    register_task_status,
+    task_status_manager,
+    update_task_status,
+)
+
+
+def test_task_status():
+    """test task status registration and update using the utils.task_status module."""
+    # Check task status registration
+    cancel_callback_mock = Mock()
+    task_status_id = register_task_status(
+        'test-task-status',
+        Status.BUSY,
+        'Register task status busy',
+        cancel_callback=cancel_callback_mock,
+    )
+    assert task_status_manager.is_busy()
+    assert task_status_manager.get_status() == [
+        'test-task-status: Register task status busy'
+    ]
+
+    # Check task status update
+    update_task_status(
+        task_status_id, Status.DONE, description='Register task status done'
+    )
+    assert not task_status_manager.is_busy()
+    assert task_status_manager.get_status() == []
+    update_task_status(
+        task_status_id,
+        Status.PENDING,
+        description='Register task status pending',
+    )
+    assert task_status_manager.is_busy()
+    assert task_status_manager.get_status() == [
+        'test-task-status: Register task status pending'
+    ]
+
+    # Check cancel behavior
+    task_status_manager.cancel_all()
+    cancel_callback_mock.assert_called_once()
+    assert task_status_manager.get_status() == []
+    assert not task_status_manager.is_busy()

src/napari/utils/task_status.py

@@ -0,0 +1,152 @@
+import datetime
+import uuid
+from enum import auto
+from typing import Optional
+
+from napari.utils.misc import Callable, StringEnum
+
+
+class Status(StringEnum):
+    PENDING = auto()
+    BUSY = auto()
+    DONE = auto()
+    CANCELLED = auto()
+    FAILED = auto()
+
+
+class TaskStatusItem:
+    def __init__(
+        self,
+        provider: str,
+        status: Status,
+        description: str,
+        cancel_callback: Optional[Callable] = None,
+    ) -> None:
+        self.id: uuid.UUID = uuid.uuid4()
+        self._provider = provider
+        self._timestamp = [self._timestap()]
+        self._status = [status]
+        self._description = [description]
+        self._cancel_callback = cancel_callback
+
+    def _timestap(self) -> str:
+        return datetime.datetime.now().isoformat()
+
+    def __str__(self) -> str:
+        return f'TaskStatusItem: ({self._provider}, {self.id}, {self._timestamp[-1]}, {self._status[-1]}, {self._description[-1]})'
+
+    def update(self, status: Status, description: str) -> None:
+        self._timestamp.append(self._timestap())
+        self._status.append(status)
+        self._description.append(description)
+
+    def cancel(self) -> bool:
+        self.update(Status.CANCELLED, '')
+        if self._cancel_callback is not None:
+            return self._cancel_callback()
+        return False
+
+    def state(self) -> tuple[str, str, Status, str]:
+        return (
+            self._provider,
+            self._timestamp[-1],
+            self._status[-1],
+            self._description[-1],
+        )
+
+
+class TaskStatusManager:
+    """
+    A task status manager, to store status of long running processes/tasks.
+
+    Only one instance is in general available through napari.
+
+    napari methods and plugins can use it to register and update
+    long running tasks.
+    """
+
+    _tasks: dict[uuid.UUID, TaskStatusItem]
+
+    def __init__(self) -> None:
+        # Note: we are using a dict here that may not be thread-safe; however
+        # given the that the values from it are added/updated using an UUID
+        # collision chances are low and it should be ok as long as operations
+        # that require its iteration (`is_busy`, `get_status`, `cancel_all`)
+        # are done when no task status additions are scheduled (i.e when
+        # closing the application).
+        self._tasks: dict[uuid.UUID, TaskStatusItem] = {}
+
+    def register_task_status(
+        self,
+        provider: str,
+        task_status: Status,
+        description: str,
+        cancel_callback: Optional[Callable] = None,
+    ) -> uuid.UUID:
+        item = TaskStatusItem(
+            provider, task_status, description, cancel_callback
+        )
+        self._tasks[item.id] = item
+        return item.id
+
+    def update_task_status(
+        self,
+        status_id: uuid.UUID,
+        task_status: Status,
+        description: str = '',
+    ) -> bool:
+        if status_id in self._tasks:
+            item = self._tasks[status_id]
+            item.update(task_status, description)
+            return True
+
+        return False
+
+    def is_busy(self) -> bool:
+        for _, item in self._tasks.items():
+            if item.state()[2] in [Status.PENDING, Status.BUSY]:
+                return True
+        return False
+
+    def get_status(self) -> list[str]:
+        messages = []
+        for _, item in self._tasks.items():
+            provider, ts, status, description = item.state()
+            if status in [Status.PENDING, Status.BUSY]:
+                messages.append(f'{provider}: {description}')
+
+        return messages
+
+    def cancel_all(self) -> None:
+        for _, item in self._tasks.items():
+            item.cancel()
+
+
+task_status_manager = TaskStatusManager()
+
+
+def register_task_status(
+    provider: str,
+    task_status: Status,
+    description: str,
+    cancel_callback: Optional[Callable] = None,
+) -> uuid.UUID:
+    """
+    Register a long running task.
+    """
+    return task_status_manager.register_task_status(
+        provider, task_status, description, cancel_callback
+    )
+
+
+def update_task_status(
+    task_status_id: uuid.UUID,
+    status: Status,
+    description: str = '',
+) -> bool:
+    """
+    Update a long running task.
+    """
+    return task_status_manager.update_task_status(
+        task_status_id, status, description
+    )