tasks.py

"""Small QThreadPool facade with per-key de-duplication."""

from __future__ import annotations

import logging
from collections.abc import Callable
from typing import Any

from PySide6.QtCore import QObject, QRunnable, QThread, QThreadPool, Signal, Slot

log = logging.getLogger(__name__)


class _TaskSignals(QObject):
    done = Signal(str, object)
    failed = Signal(str, str)


class _Task(QRunnable):
    def __init__(self, key: str, work: Callable[[], Any]) -> None:
        super().__init__()
        self.key = key
        self._work = work
        self.signals = _TaskSignals()
        self.setAutoDelete(True)

    def run(self) -> None:
        try:
            result = self._work()
        except Exception as exc:  # boundary: never let a worker die silently
            log.exception("task %r failed", self.key)
            self.signals.failed.emit(self.key, str(exc))
        else:
            self.signals.done.emit(self.key, result)


class TaskRunner(QObject):
    """Runs blocking WinAPI work off the GUI thread.

    ``key`` identifies the logical operation; submitting the same key while
    it is still running is ignored, which prevents retry storms and
    double-clicks from spawning duplicate threads.
    """

    def __init__(self, parent: QObject | None = None) -> None:
        super().__init__(parent)
        self._pool = QThreadPool(self)
        self._pool.setMaxThreadCount(max(2, min(8, QThread.idealThreadCount())))
        self._callbacks: dict[str, tuple[Callable | None, Callable | None]] = {}
        self._tasks: dict[str, _Task] = {}

    def submit(
        self,
        key: str,
        work: Callable[[], Any],
        on_done: Callable[[Any], None] | None = None,
        on_failed: Callable[[str], None] | None = None,
    ) -> bool:
        if key in self._callbacks:
            log.debug("task %r already running, skipped", key)
            return False

        task = _Task(key, work)
        task.signals.done.connect(self._on_done)
        task.signals.failed.connect(self._on_failed)
        self._callbacks[key] = (on_done, on_failed)
        self._tasks[key] = task
        self._pool.start(task)
        return True

    def is_running(self, key: str) -> bool:
        return key in self._callbacks

    def wait(self, msec: int = 3000) -> bool:
        return self._pool.waitForDone(msec)

    @Slot(str, object)
    def _on_done(self, key: str, result: object) -> None:
        callbacks = self._callbacks.pop(key, None)
        self._tasks.pop(key, None)
        if callbacks and callbacks[0] is not None:
            callbacks[0](result)

    @Slot(str, str)
    def _on_failed(self, key: str, message: str) -> None:
        callbacks = self._callbacks.pop(key, None)
        self._tasks.pop(key, None)
        if callbacks and callbacks[1] is not None:
            callbacks[1](message)