PyQt5 QThread with QObject example

qthread_example.py

import sys

from PyQt5.QtCore import pyqtSlot, QThread, QObject, pyqtSignal
from PyQt5.QtWidgets import QMainWindow, QApplication


class Worker(QObject):
    finished = pyqtSignal()

    def __init__(self):
        super(Worker, self).__init__()

        self._is_stop_requested: bool = False

    @pyqtSlot()
    def request_stop(self):
        """ Emit to this slot to stop worker execution early. """
        self._is_stop_requested = True

    @pyqtSlot()
    def run(self):
        for i in range(1000000):
            if self._is_stop_requested:
                break
            if i % 1000 == 0:
                print(f"{i}")
        self.finished.emit()


class MyApp(QMainWindow):
    def __init__(self, parent=None):
        super().__init__(parent)

        # Init worker and thread.
        self.worker = Worker()
        self.thread = QThread()
        # Move worker to thread.
        # This must be done before attaching any signals to the worker.
        self.worker.moveToThread(self.thread)

        # Attach signals
        self.thread.started.connect(self.worker.do_work)
        self.thread.finished.connect(self.on_thread_finished)
        self.thread.finished.connect(self.thread.deleteLater)

        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)

        # Start thread
        self.thread.start()

    @pyqtSlot()
    def on_thread_finished(self):
        print("Thread finished.")

    @pyqtSlot()
    def on_worker_finished(self):
        print("Worker finished.")
        self.thread.quit()
        self.thread.wait()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    example = MyApp()
    sys.exit(app.exec_())