PyQt5: calling widget slots safely from a worker thread

Question

How to call widget slots from QThread worker? I know I could create a signal for each widget's slot like this:

class App(QtWidgets.QMainWindow):
    signal_line_edit_1_setText = pyqtSignal(str)
    signal_line_edit_2_setText = pyqtSignal(str)
    ...

    def __init__(self):
        ...
        self.signal_line_edit_1_setText.connect(self.line_edit_1.setText)
        self.signal_line_edit_2_setText.connect(self.line_edit_2.setText)
        self.worker = Worker(self)


class Worker(QThread):
    #  Maybe I have to create signals for Worker class and then connect them to app's signals,
    #  but that would be even more complicated

    def __init__(self, app):
        self.app = app
        ...
    def run(self):
        self.app.signal_line_edit_1_setText.emit('Worker Running')
        ...

Isn't there a more simple way to thread-safely interact with widgets? QTimer works without signal wrapping, but it makes UI just a little laggy. I know about QThreadPool, but don't really understand it.

Answer 1

There is a way to do exactly what you want, but it's much less commonly used than custom signals (in Python anyway - I'm not so sure about C++).

The relevant API is QMetaObject.invokeMethod. This permits thread-safe calling of any method of a QObject subclass, so long as the method is accessible via the Qt Meta Object System. In practise, this will usually limit it to pre-existing Qt methods, plus user-defined methods wrapped with the pyqtSlot decorator. Here is what a typical usage looks like:

QtCore.QMetaObject.invokeMethod(widget, 'mySlot', QtCore.Q_ARG(int, number))

As you can see, it looks quite clunky in comparison with PyQt's new-style signal and slot syntax, and in fact its disadvantages are similar to those of the old-style signal and slot syntax: namely, that it's somewhat error-prone, verbose and not very pythonic. Its only significant advantage (relative to the current use-case) is that it avoids having to pre-define a signal. (See below for a demo script that uses both approaches to ensure GUI updates are carried out in the main thread).

I suppose it would be possible to create a custom class that automagically invoked methods across threads using this approach (perhaps via __getattr__). But why bother going to all the trouble of developing and maintaining such a class when there's already a built-in mechanism that achieves much the same thing? Defining a custom signal, connecting it to a callable, and emitting it isn't at all complicated:

class Worker(QThread):
    customSignal = pyqtSignal(int)

    def run(self):
        self.customSignal.emit(42)

worker = Worker()
worker.customSignal.connect(lambda x: print(x))
worker.start()

and the resulting code is very readable, flexible and easy to maintain.

---

Demo Script:

from PyQt5 import QtCore, QtWidgets

def thread_id():
    return int(QtCore.QThread.currentThreadId())

class Worker(QtCore.QThread):
    progressChanged = QtCore.pyqtSignal(int)

    def setMethod(self, invoke=False):
        self._invoke = invoke

    def run(self):
        print()
        print(f'Thread: {MAIN_THREAD} [Main]')
        print(f'Thread: {thread_id()} [Worker.run]')
        invoke = getattr(self, '_invoke', False)
        print('Using Method:', 'invoke' if invoke else 'signal')
        for count in range(1, 6):
            self.msleep(500)
            if invoke:
                QtCore.QMetaObject.invokeMethod(
                    self.parent(), 'updateProgress', QtCore.Q_ARG(int, count))
            else:
                self.progressChanged.emit(count)

class Window(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.button = QtWidgets.QPushButton('Test')
        self.button.clicked.connect(self.handleButton)
        self.check = QtWidgets.QCheckBox('Use inkoke')
        self.label = QtWidgets.QLabel()
        self.label.setAlignment(QtCore.Qt.AlignCenter)
        layout = QtWidgets.QGridLayout(self)
        layout.addWidget(self.label, 0, 0, 1, 2)
        layout.addWidget(self.button, 1, 0)
        layout.addWidget(self.check, 1, 1)
        self.worker = Worker(self)
        self.worker.progressChanged.connect(self.updateProgress)
        self.updateProgress()

    def handleButton(self):
        if not self.worker.isRunning():
            self.updateProgress()
            self.worker.setMethod(invoke=self.check.isChecked())
            self.worker.start()

    @QtCore.pyqtSlot(int)
    def updateProgress(self, count=0):
        if count:
            print(f'Thread: {MAIN_THREAD} [Main]')
            print(f'Thread: {thread_id()} [Window.updateProgress]')
        self.label.setText(f'Count: {count}')

    def closeEvent(self, event):
        self.worker.quit()
        self.worker.wait()


app = QtWidgets.QApplication(['Test'])

MAIN_THREAD = thread_id()
print(f'Thread: {MAIN_THREAD} [Main]')

window = Window()
window.setGeometry(600, 100, 300, 200)
window.show()

app.exec_()