Source code for forkbit_sdk.worker

from __future__ import annotations

import logging
import traceback
from typing import TYPE_CHECKING

from PySide6.QtCore import QThread, Signal

if TYPE_CHECKING:
    from forkbit_sdk.base_plugin import BasePlugin

log = logging.getLogger(__name__)




class PluginWorker(QThread):
    """Base class for plugin background tasks.

    Handles busy-state tracking, logging, and error handling automatically.
    Subclass and override :meth:`execute` with your work.

    Example::

        class BuildWorker(PluginWorker):
            def execute(self) -> str:
                self.log_message.emit("Building...")
                # do work
                return "Build complete"

        # In your plugin:
        worker = BuildWorker(self)
        worker.log_message.connect(self._append_log)
        worker.done.connect(self._on_done)
        worker.start()

    The plugin's busy indicator is set automatically when the worker
    starts and cleared when it finishes — no need to call
    :meth:`~BasePlugin.set_busy` yourself.
    """

    log_message = Signal(str)
    """Emitted to log a message to the plugin console."""

    step_changed = Signal(int)
    """Emitted to advance the pipeline step indicator."""

    progress = Signal(int, int)
    """Emitted with (current, total) for progress tracking."""

    done = Signal(bool, str)
    """Emitted when the worker finishes: (success, message)."""

    _busy_signal = Signal(bool)
    """Internal signal to update busy state thread-safely."""



    def __init__(self, plugin: BasePlugin):
        super().__init__()
        self._plugin = plugin
        self._busy_signal.connect(plugin.set_busy)




    def start(self, priority=QThread.Priority.InheritPriority) -> None:
        """Start the worker thread and mark the plugin as busy."""
        self._plugin.set_busy(True)
        super().start(priority)




    def run(self) -> None:
        try:
            result = self.execute()
            self.done.emit(True, result or "")
        except Exception as e:
            log.warning("Worker failed: %s", e, exc_info=True)
            self.log_message.emit(traceback.format_exc())
            self.done.emit(False, str(e))
        finally:
            self._busy_signal.emit(False)




    def execute(self) -> str:
        """Override this to do the actual work.

        Use :attr:`log_message`, :attr:`step_changed`, and :attr:`progress`
        signals to report status. Return a success message string.

        :returns: A message shown on completion (e.g. ``"Build complete"``).
        :raises Exception: Any exception is caught and reported as failure.
        """
        raise NotImplementedError