AsyncTask

asynctask.py

class AsyncTask(QRunnable):
    class _Signals(QObject):
        result = pyqtSignal(tuple)

        @staticmethod
        def _callback(args: Tuple[Callable[[Any], None], Any]):
            fun, arg = args
            fun(arg)

        def __init__(self):
            super().__init__()
            self.result.connect(self._callback)

    _signals = _Signals()

    def _default_callback(*arg, **kwargs):
        pass

    def __init__(self,
                 routine: Callable[[], Any],
                 on_result: Callable[[Any], None] = None,
                 on_error: Callable[[Exception], None] = None):
        super().__init__()

        if on_result is None:
            on_result = self._default_callback
        self._on_result = on_result

        if on_error is None:
            on_error = self._default_callback
        self._on_error = on_error

        self._routine = routine

    def run(self):
        try:
            result = self._routine()
            self._signals.result.emit((self._on_result, result))
        except Exception as ex:
            self._signals.result.emit((self._on_error, ex))

    def start(self, thread_pool: QThreadPool = None):
        if thread_pool is None:
            QThreadPool.globalInstance().start(self)
        else:
            thread_pool.start(self)