searx-qt/searxqt/thread.py

########################################################################
#  Searx-Qt - Lightweight desktop application for Searx.
#  Copyright (C) 2020-2022  CYBERDEViL
#
#  This file is part of Searx-Qt.
#
#  Searx-Qt is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  Searx-Qt is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
########################################################################

from PyQt5.QtCore import pyqtSignal, QThread, QObject


class ThreadManagerProto(QObject):
    threadStarted = pyqtSignal()
    threadFinished = pyqtSignal()

    def __init__(self, parent=None):
        QObject.__init__(self, parent=parent)
        self._thread = None
        self._threadQueue = []

    def cancelAll(self):
        self._threadQueue.clear()
        self._thread.wait()

    def currentJobStr(self):
        return ""

    def queueCount(self):
        return len(self._threadQueue)

    def hasActiveJobs(self):
        return bool(
            self._thread or self._threadQueue
        )


class Thread(QThread):
    finished = pyqtSignal()

    def __init__(self, func, args=None, kwargs=None, parent=None):
        QThread.__init__(self, parent=parent)
        self._func = func
        self._args = args
        self._kwargs = kwargs

        self._result = None

    def result(self): return self._result

    def run(self):
        if self._args and self._kwargs:
            self._result = self._func(*self._args, **self._kwargs)
        elif self._args:
            self._result = self._func(*self._args)
        elif self._kwargs:
            self._result = self._func(**self._kwargs)
        else:
            self._result = self._func()
        self.finished.emit()