32. Signals & Slots Connection Types
32.1 Direct Connection
1 import sys
2 from PySide6.QtCore import Signal, Slot, QEvent
3 from PySide6.QtWidgets import (QApplication,
4 QWidget, QPushButton, QVBoxLayout)
5
6
7 class Window(QWidget):
8
9 # 1. Create a custom signal
10
11 custom_signal = Signal()
12
13 def __init__(self):
14
15 super().__init__()
16
17 layout = QVBoxLayout()
18 self.setLayout(layout)
19
20 self.button = QPushButton('Click me!')
21 layout.addWidget(self.button)
22
23 self.button.clicked.connect(self.on_button_clicked)
24
25 # 3. Connect the signal with the slots
26
27 self.custom_signal.connect(self.on_custom_signal_1)
28 self.custom_signal.connect(self.on_custom_signal_2)
29 self.custom_signal.connect(self.on_custom_signal_3)
30
31 # 4. Emit the signal
32
33 @Slot()
34 def on_button_clicked(self):
35 print('on_button_clicked: Emitting custom signal\n')
36 self.custom_signal.emit()
37 print('on_button_clicked: Returning')
38
39 # 2. Create the slots
40
41 @Slot()
42 def on_custom_signal_1(self):
43 print('First custom slot executed')
44
45 @Slot()
46 def on_custom_signal_2(self):
47 print('Second custom slot executed')
48
49 @Slot()
50 def on_custom_signal_3(self):
51 print('Third custom slot executed\n')
52
53 def event(self, event):
54 if event.type() == QEvent.Type.MetaCall:
55 print(f"Queued signal (MetaCallEvent) intercepted")
56 return QWidget.event(self, event)
57
58
59 if __name__ == '__main__':
60
61 app = QApplication(sys.argv)
62
63 main_window = Window()
64 main_window.show()
65
66 sys.exit(app.exec())
With Qt.DirectConnection the slot is invoked immediately when the signal is emitted. In the example we create a custom signal named custom_signal. We also create three slots, on_custom_signal_1, on_custom_signal_2 and on_custom_signal_3 and connect the signal with all three slots. We add a button to the main window and emit custom_signal when it is clicked.
The output is
1 on_button_clicked: Emitting custom signal
2
3 First custom slot executed
4 Second custom slot executed
5 Third custom slot executed
6
7 on_button_clicked: Returning
From the output we see that when we emit the signal in on_button_clicked the three slots are called sequentially and, only after on_custom_signal_3 completes and returns, on_button_clicked continues execution. The code execution is synchronous.
When establishing the connections we didn’t pass the optional type to the Signal.connect() method. type defaults to Qt.AutoConnection and since both the signal and the slots live in the main thread the connection becomes Qt.DirectConnection.
There is one more thing to note: The slot execution bypasses the Qt event loop. We reimplement Window.event() to print a message but it is never executed.
32.2 Queued Connection
We have the same setup as in the direct connection example: a single custom signal connected to three slots.
1 import sys
2 from PySide6.QtCore import Signal, Slot, QEvent, Qt
3 from PySide6.QtWidgets import (QApplication,
4 QWidget, QPushButton, QVBoxLayout)
5
6
7 class Window(QWidget):
8
9 custom_signal = Signal()
10
11 def __init__(self):
12
13 super().__init__()
14
15 layout = QVBoxLayout()
16 self.setLayout(layout)
17
18 self.button = QPushButton('Click me!')
19 layout.addWidget(self.button)
20
21 self.button.clicked.connect(self.on_button_clicked)
22
23 # 1. Force the queued connection
24
25 self.custom_signal.connect(self.on_custom_signal_1,
26 Qt.ConnectionType.QueuedConnection)
27 self.custom_signal.connect(self.on_custom_signal_2,
28 Qt.ConnectionType.QueuedConnection)
29 self.custom_signal.connect(self.on_custom_signal_3,
30 Qt.ConnectionType.QueuedConnection)
31
32 @Slot()
33 def on_button_clicked(self):
34 print('on_button_clicked: Emitting custom signal')
35 self.custom_signal.emit()
36 print('on_button_clicked: Returning\n')
37
38 @Slot()
39 def on_custom_signal_1(self):
40 print('First custom slot executed\n')
41
42 @Slot()
43 def on_custom_signal_2(self):
44 print('Second custom slot executed\n')
45
46 @Slot()
47 def on_custom_signal_3(self):
48 print('Third custom slot executed\n')
49
50 # 2. Intercept MetaCall events
51
52 def event(self, event):
53 if event.type() == QEvent.Type.MetaCall:
54 print(f"Queued signal (MetaCall event) intercepted")
55 return QWidget.event(self, event)
56
57
58 if __name__ == '__main__':
59
60 app = QApplication(sys.argv)
61
62 main_window = Window()
63 main_window.show()
64
65 sys.exit(app.exec())
But now the output is
1 on_button_clicked: Emitting custom signal
2 on_button_clicked: Returning
3
4 Queued signal (MetaCallEvent) intercepted
5 First custom slot executed
6
7 Queued signal (MetaCallEvent) intercepted
8 Second custom slot executed
9
10 Queued signal (MetaCallEvent) intercepted
11 Third custom slot executed
Now each slot execution is queued in the Qt event loop. After emitting the signal on_button_clicked continues execution and returns immediately, returning control to the event loop, allowing the slots to be executed. Each slot execution is wrapped in an asynchronous method invocation via QMetaObject::invokeMethod().
All three slots are still executed in the receiver thread which happens to be the main thread.
32.3 Blocking Queued Connection
Qt.BlockingQueuedConnection behaves the same as Qt::QueuedConnection, except that the signalling thread blocks until the slot returns.
1 from PySide6.QtCore import QObject, Signal, Slot, QThread
2
3 class Sender(QObject):
4
5 operate = Signal()
6
7 @Slot()
8 def handle_results(self):
9 print('Sender: handle_results started')
10 QThread.sleep(5)
11 print('Sender: handle_results finished after delay')
1 from PySide6.QtCore import QObject, Signal, Slot
2
3 class Receiver(QObject):
4
5 result_ready = Signal()
6
7 @Slot()
8 def do_work(self):
9 print('Receiver: Starting do_work')
10 print('Receiver: Emitting result_ready')
11 self.result_ready.emit()
12 print('Receiver: After emit (this should delay if blocking)')
We have two worker objects: the Sender object signals the Receiver to do_work. Receiver in turn signals to the Sender that the work results are ready so that the Sender can handle_results.
1 import sys
2 from PySide6.QtCore import QThread, Slot, Qt
3 from PySide6.QtWidgets import (QApplication,
4 QPushButton, QWidget, QVBoxLayout)
5 from sender import Sender
6 from receiver import Receiver
7
8
9 class Window(QWidget):
10
11 def __init__(self):
12
13 super().__init__()
14
15 layout = QVBoxLayout()
16 self.setLayout(layout)
17
18 self.button = QPushButton('Start the task')
19 self.button.clicked.connect(self.on_button_clicked)
20 layout.addWidget(self.button)
21
22 self.sender_thread = QThread()
23 self.receiver_thread = QThread()
24
25 self.sender_obj = Sender()
26 self.receiver_obj = Receiver()
27
28 self.sender_obj.moveToThread(self.sender_thread)
29 self.receiver_obj.moveToThread(self.receiver_thread)
30
31 self.sender_thread.finished.connect(
32 self.sender_obj.deleteLater)
33 self.receiver_thread.finished.connect(
34 self.receiver_obj.deleteLater)
35
36 self.sender_obj.operate.connect(self.receiver_obj.do_work)
37
38 self.receiver_obj.result_ready.connect(
39 self.sender_obj.handle_results,
40 Qt.ConnectionType.BlockingQueuedConnection)
41 '''
42 self.receiver_obj.result_ready.connect(
43 self.sender_obj.handle_results)
44 '''
45
46 self.sender_thread.start()
47 self.receiver_thread.start()
48
49 @Slot()
50 def on_button_clicked(self):
51 self.sender_obj.operate.emit()
52
53 def closeEvent(self, event):
54 try:
55 self.sender_thread.quit()
56 self.receiver_thread.quit()
57 self.sender_thread.wait()
58 self.receiver_thread.wait()
59 except Exception as e:
60 print(e)
61 event.accept()
62
63
64 if __name__ == '__main__':
65
66 app = QApplication(sys.argv)
67
68 main_window = Window()
69 main_window.show()
70
71 sys.exit(app.exec())
In the main window we connect the signals with the slots. If we use Qt.BlockingQueuedConnection:
1 self.receiver_obj.result_ready.connect(
2 self.sender_obj.handle_results,
3 Qt.ConnectionType.BlockingQueuedConnection)
the output is
1 Receiver: Starting do_work
2 Receiver: Emitting result_ready
3 Sender: handle_results started
4 Sender: handle_results finished after delay
5 Receiver: After emit (this should delay if blocking)
The Receiver thread is blocked for five seconds after it emits results_ready and only prints the last message after Sender finished handling the results.
If we used QtQueuedConnection instead
1 self.receiver_obj.result_ready.connect(
2 self.sender_obj.handle_results)
the output is
1 Receiver: Starting do_work
2 Receiver: Emitting result_ready
3 Receiver: After emit (this should delay if blocking)
4 Sender: handle_results started
5 Sender: handle_results finished after delay
which means that do_work emits the results_ready signal and immediately continues its execution.
32.4 Unique Connection
Making your signal-slot connections unique is a good idea most of the time and Qt.UniqueConnection lets you make that explicit.
1 import sys
2 from PySide6.QtCore import Slot, Qt, SIGNAL
3 from PySide6.QtWidgets import (QApplication, QWidget,
4 QHBoxLayout, QVBoxLayout, QLabel, QPushButton)
5
6
7 class Window(QWidget):
8
9 def __init__(self):
10 super().__init__()
11
12 self.resize(300, 200)
13 layout = QVBoxLayout()
14 self.setLayout(layout)
15
16 self.button = QPushButton('Click Me')
17 self.label = QLabel('Button: 0 receivers')
18
19 layout.addWidget(self.button)
20 layout.addWidget(self.label)
21
22 h_layout = QHBoxLayout()
23 self.connect_btn = QPushButton('Connect signals')
24 self.disconnect_btn = QPushButton('Disconnect signals')
25
26 self.connect_btn.clicked.connect(
27 self.on_connect_btn_clicked)
28 self.disconnect_btn.clicked.connect(
29 self.on_disconnect_btn_clicked)
30
31 h_layout.addWidget(self.connect_btn)
32 h_layout.addWidget(self.disconnect_btn)
33 layout.addLayout(h_layout)
34
35 def on_connect_btn_clicked(self):
36 conn = self.button.clicked.connect(self.on_clicked,
37 Qt.ConnectionType.UniqueConnection)
38 if conn:
39 print('Connection is valid')
40 else:
41 print('Connection is invalid')
42 self.update_label()
43
44 def on_disconnect_btn_clicked(self):
45 self.button.clicked.disconnect(self.on_clicked)
46 self.update_label()
47
48 @Slot(bool)
49 def on_clicked(self, checked):
50 print('Button clicked')
51
52 def update_label(self):
53 count = self.button.receivers(SIGNAL('clicked(bool)'))
54 self.label.setText(f'Button: {count} receivers')
55
56
57 if __name__ == '__main__':
58
59 app = QApplication(sys.argv)
60 main_window = Window()
61 main_window.show()
62 sys.exit(app.exec())
The example starts without any connections made. We make the connection on clicking the connect_btn button but we use QtUniqueConnection so the connection is established only once. All subsequent attempts to connect self.button.clicked with self.on_clicked return invalid connection objects.The label text and the console output also reflect this.
1 Connection is valid
2 Connection is invalid
3 Connection is invalid
4 Connection is invalid
5 Connection is invalid
32.5 Single-Shot Connection
While Qt.UniqueConnection makes connections unique, Qt.SingleShotConnection makes them disposable: the connection is automatically broken when the signal is emitted.
1 import sys
2 from PySide6.QtCore import Slot, Qt, SIGNAL
3 from PySide6.QtWidgets import (QApplication, QWidget,
4 QVBoxLayout, QLabel, QPushButton)
5
6
7 class Window(QWidget):
8
9 def __init__(self):
10 super().__init__()
11
12 self.resize(300, 200)
13 layout = QVBoxLayout()
14 self.setLayout(layout)
15
16 self.button = QPushButton('Click Me')
17 self.connect_btn = QPushButton('Connect signals')
18 self.label = QLabel('Button: 0 receivers')
19
20 layout.addWidget(self.button)
21 layout.addWidget(self.connect_btn)
22 layout.addWidget(self.label)
23
24 self.connect_btn.clicked.connect(
25 self.on_connect_btn_clicked)
26
27 def on_connect_btn_clicked(self):
28 self.button.clicked.connect(self.on_clicked,
29 Qt.ConnectionType.SingleShotConnection)
30 self.update_label()
31
32 @Slot(bool)
33 def on_clicked(self, checked):
34 print('Button clicked')
35 self.update_label()
36
37 def update_label(self):
38 count = self.button.receivers(SIGNAL('clicked(bool)'))
39 self.label.setText(f'Button: {count} receivers')
40
41
42 if __name__ == '__main__':
43
44 app = QApplication(sys.argv)
45 main_window = Window()
46 main_window.show()
47 sys.exit(app.exec())
The example lets you accumulate one-shot connections by clicking the Connect signals button and their count is displayed in the label. Once you click the Click Me button all connections are broken and the on_clicked slot is executed as many times as there were connections.