9d4e0995fd
The reason is that `sys.excepthook` is a global attribute which we have to unset for threads because we show a Qt dialog in it and we can't do that from threads. Before this change, we used to unset it in threads, but since it is a global attr, that was unsetting it for the whole application. We cannot reliably reset it because we can have multiple threads executing and there will be race conditions. To fix this situation, `RareAppException` implements a callback to be patched into `sys.excepthook` which emits a signal to be serviced by the the `RareAppException` instance in the main thread. `RareAppException` can be subclassed to implement the `RareAppException._handler` method for domain specific handling. The `RareApp` base class instantiates its own `RareAppException` instance for early basic handling. `RareAppException` is subclassed into `RareException` and `RareLauncherExcpetion` in `Rare` and `RareLauncher` respectively to implement the aforemention domain specific handling. Each of these classes deletes the previous instance and replace it with their specialized handlers.
93 lines
2.2 KiB
Python
93 lines
2.2 KiB
Python
import sys
|
|
from abc import abstractmethod
|
|
from dataclasses import dataclass
|
|
from enum import IntEnum
|
|
from typing import Optional
|
|
|
|
from PyQt5.QtCore import QRunnable, QObject, pyqtSlot, pyqtSignal
|
|
|
|
|
|
class Worker(QRunnable):
|
|
"""
|
|
Base QRunnable class.
|
|
|
|
This class provides a base for QRunnables with signals that are automatically deleted.
|
|
|
|
To use this class you have to assign the signals object of your concrete implementation
|
|
to the `Worker.signals` attribute and implement `Worker.run_real()`
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(Worker, self).__init__()
|
|
self.setAutoDelete(True)
|
|
self.__signals: Optional[QObject] = None
|
|
|
|
@property
|
|
def signals(self) -> QObject:
|
|
if self.__signals is None:
|
|
raise NotImplementedError
|
|
return self.__signals
|
|
|
|
@signals.setter
|
|
def signals(self, obj: QObject):
|
|
self.__signals = obj
|
|
|
|
@abstractmethod
|
|
def run_real(self):
|
|
pass
|
|
|
|
@pyqtSlot()
|
|
def run(self):
|
|
self.run_real()
|
|
self.__signals.deleteLater()
|
|
|
|
|
|
class QueueWorkerState(IntEnum):
|
|
UNDEFINED = 0
|
|
QUEUED = 1
|
|
ACTIVE = 2
|
|
|
|
|
|
@dataclass
|
|
class QueueWorkerInfo:
|
|
app_name: str
|
|
app_title: str
|
|
worker_type: str
|
|
state: QueueWorkerState
|
|
progress: int = 0
|
|
|
|
|
|
class QueueWorker(Worker):
|
|
"""
|
|
Base queueable worker class
|
|
|
|
This class is a specialization of the `Worker` class. It provides feedback signals to know
|
|
if a worker has started or finished.
|
|
|
|
To use this class you have to assign the signals object of your concrete implementation
|
|
to the `QueueWorker.signals` attribute, implement `QueueWorker.run_real()` and `QueueWorker.worker_info()`
|
|
"""
|
|
|
|
class Signals(QObject):
|
|
started = pyqtSignal()
|
|
finished = pyqtSignal()
|
|
|
|
def __init__(self):
|
|
super(QueueWorker, self).__init__()
|
|
self.feedback = QueueWorker.Signals()
|
|
self.state = QueueWorkerState.QUEUED
|
|
|
|
@pyqtSlot()
|
|
def run(self):
|
|
self.state = QueueWorkerState.ACTIVE
|
|
self.feedback.started.emit()
|
|
super(QueueWorker, self).run()
|
|
self.feedback.finished.emit()
|
|
self.feedback.deleteLater()
|
|
|
|
@abstractmethod
|
|
def worker_info(self) -> QueueWorkerInfo:
|
|
pass
|
|
|
|
|