I am working on a PySide6 project and I have issue. I have a simple GUI with a button that triggers a function but the UI completely freezes while the function is running. I know this is because the function is blocking the main thread but I am not sure of the best way to fix it.
I have read a bit about using QThread or QThreadPool but honestly, threading always confuses me. I also tried moving the function to a separate thread but now I am facing issues where the UI does not update properly or the thread does not stop as expected.
What is the best approach to running a long task in PySide without freezing the UI? Would QThreadPool be better than QThread or is there an easier way to do this? I came across a Java Course Online while searching for solutions but I want to hear from someone with hands-on PySide experience.
You can use QThreadPool with QRunnable, which is a high level interface for threading. Then you can use Signals and Slots to communicate between you UI thread and your separate thread for keeping your UI responsive and up to date.
Here is a simple example. Note that you need to inherit both from QObject and QRunnable in your implementation. QObject is for Signals and Slots.
import sys
import time
from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import (
QApplication,
QWidget,
QPushButton,
QVBoxLayout,
QProgressBar,
QLabel,
)
class LongTask(QObject, QRunnable):
finished = Signal(str)
error = Signal(str)
progress = Signal(int)
def __init__(self, task_id, parent=None):
QObject.__init__(self, parent)
QRunnable.__init__(self)
self.task_id = task_id
self._is_cancelled = False
@Slot()
def cancel(self):
self._is_cancelled = True
def run(self):
try:
for i in range(101):
if self._is_cancelled:
self.error.emit(f"Task {self.task_id} cancelled.")
return
time.sleep(0.05) # Simulate a long task
self.progress.emit(i)
self.finished.emit(f"Task {self.task_id} completed.")
except Exception as e:
self.error.emit(f"Task {self.task_id} error: {e}")
class MainWindow(QWidget):
def __init__(self):
super().__init__()
self.threadpool = QThreadPool()
self.threadpool.setMaxThreadCount(2) # Limit threads
self.start_button = QPushButton("Start Task")
self.cancel_button = QPushButton("Cancel Task")
self.cancel_button.setEnabled(False)
self.progress_bar = QProgressBar()
self.progress_bar.setValue(0)
self.result_label = QLabel("")
layout = QVBoxLayout()
layout.addWidget(self.start_button)
layout.addWidget(self.cancel_button)
layout.addWidget(self.progress_bar)
layout.addWidget(self.result_label)
self.setLayout(layout)
self.start_button.clicked.connect(self.start_task)
self.cancel_button.clicked.connect(self.cancel_task)
self.task = None
def start_task(self):
self.start_button.setEnabled(False)
self.cancel_button.setEnabled(True)
self.progress_bar.setValue(0)
self.result_label.setText("Task started...")
self.task = LongTask("Task1") # Unique ID
self.task.finished.connect(self.task_finished)
self.task.error.connect(self.task_error)
self.task.progress.connect(self.task_progress)
self.threadpool.start(self.task)
def cancel_task(self):
if self.task:
self.task.cancel()
def task_finished(self, result):
self.result_label.setText(result)
self.start_button.setEnabled(True)
self.cancel_button.setEnabled(False)
def task_error(self, error_message):
self.result_label.setText(error_message)
self.start_button.setEnabled(True)
self.cancel_button.setEnabled(False)
def task_progress(self, progress):
self.progress_bar.setValue(progress)
if __name__ == "__main__":
# Try getting an existing instance in case we are in a notebook
app = QApplication.instance() or QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())
