PyQt QThreadPool

Summary: in this tutorial, you’ll learn how to create a PyQt multithreading application that uses QThreadPool and QRunnable classes.

Introduction to the QThreadPool & QRunnable classes

The QThread class allows you to offload a long-running task to a worker thread to make the application more responsive. The QThread class works fine if the application has a few worker threads.

A multithreaded program is efficient when it has a number of QThread objects that correspond to the number of CPU cores.

Also, creating threads is quite expensive in terms of computer resources. Therefore, the program should reuse created threads as much as possible.

So using the QThread class to manage worker threads has two main challenges:

  • Determine the ideal number of threads for the application based on the number of CPU cores.
  • Reuse and recycle the threads as much as possible.

Fortunately, PyQt has QThreadPool class that solves these challenges for you. The QThreadPool class is often used with the QRunnable class.

  • The QRunnable class represents a task you want to execute in a worker thread.
  • The QThreadPool executes a QRunnable object, and manages and recycles threads automatically.

Each Qt application has a global QThreadPool object which can be accessed via the globalInstance() static method of the QThreadPool class.

To use the QThreadPool and QRunnable classes, you follow these steps:

First, create a class that inherits from the QRunnable class and override the run() method:

class Worker(QRunnable): @Slot() def run(self): # place a long-running task here pass
Code language: Python (python)

Second, access the thread pool from the main window and start the worker threads:

class MainWindow(QMainWindow): # other methods # ... def start(self): """ Create and execute worker threads """ pool = QThreadPool.globalInstance() for _ in range(1, 100): pool.start(Worker())
Code language: Python (python)

To update the worker’s progress to the main thread, you use signals and slots. However, the QRunnable doesn’t support signal.

Therefore, you need to define a separate class that inherits from the QObject and uses that class in the Worker class. Here are the steps:

First, define the Signals class that subclasses the QObject class:

class Signals(QObject): completed = Signal()
Code language: Python (python)

In the Signals class, we define one signal called completed. Note that you can define as many signals as needed.

Second, emit the completed signal when the job is done in the Worker class:

class Runnable(QRunnable): def __init__(self): super().__init__() self.signals = Signals() @Slot() def run(self): # long running task # ... # emit the completed signal self.signals.completed.emit()
Code language: Python (python)

Third, connect the signal of the worker thread with a slot of the main window before submitting the worker to the pool:

class MainWindow(QMainWindow): # other methods # ... def start(self): """ Create and execute worker threads """ pool = QThreadPool.globalInstance() for _ in range(1, 100): worker = Worker() worker.signals.completed.connect(self.update) pool.start(worker) def update(self): # update the worker pass
Code language: Python (python)

QThreadPool example

The following example illustrates how to use the QThreadPool and QRunnable class:

import sys import time from PyQt6.QtWidgets import QApplication, QMainWindow, QPushButton, QGridLayout, QWidget, QProgressBar, QListWidget from PyQt6.QtCore import QRunnable, QObject, QThreadPool, pyqtSignal as Signal, pyqtSlot as Slot class Signals(QObject): started = Signal(int) completed = Signal(int) class Worker(QRunnable): def __init__(self, n): super().__init__() self.n = n self.signals = Signals() @Slot() def run(self): self.signals.started.emit(self.n) time.sleep(self.n*1.1) self.signals.completed.emit(self.n) class MainWindow(QMainWindow): def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle('QThreadPool Demo') self.job_count = 10 self.comleted_jobs = [] widget = QWidget() widget.setLayout(QGridLayout()) self.setCentralWidget(widget) self.btn_start = QPushButton('Start', clicked=self.start_jobs) self.progress_bar = QProgressBar(minimum=0, maximum=self.job_count) self.list = QListWidget() widget.layout().addWidget(self.list, 0, 0, 1, 2) widget.layout().addWidget(self.progress_bar, 1, 0) widget.layout().addWidget(self.btn_start, 1, 1) self.show() def start_jobs(self): self.restart() pool = QThreadPool.globalInstance() for i in range(1, self.job_count+1): worker = Worker(i) worker.signals.completed.connect(self.complete) worker.signals.started.connect(self.start) pool.start(worker) def restart(self): self.progress_bar.setValue(0) self.comleted_jobs = [] self.btn_start.setEnabled(False) def start(self, n): self.list.addItem(f'Job #{n} started...') def complete(self, n): self.list.addItem(f'Job #{n} completed.') self.comleted_jobs.append(n) self.progress_bar.setValue(len(self.comleted_jobs)) if len(self.comleted_jobs) == self.job_count: self.btn_start.setEnabled(True) if __name__ == '__main__': app = QApplication(sys.argv) window = MainWindow() sys.exit(app.exec()) def complete(self, n): self.list.addItem(f'Job #{n} completed.') self.comleted_jobs.append(n) self.progress_bar.setValue(len(self.comleted_jobs)) if len(self.comleted_jobs) == self.job_count: self.btn_start.setEnabled(True)
Code language: Python (python)

Output:

Signals class

Define the Signals class that inherits from the QObject class to support signals. In the Signals class, we define two signals:

  • The started signal will be emitted when a worker is started.
  • The completed signal will be emitted when a worker is completed.

Both signals accept an integer that identifies the job number:

class Signals(QObject): started = Signal(int) completed = Signal(int)
Code language: Python (python)

Worker class

The Worker class inherits from the QRunnable class. The Worker class represents a long-running task that we offload to a worker thread:

class Worker(QRunnable): def __init__(self, n): super().__init__() self.n = n self.signals = Signals() @Slot() def run(self): self.signals.started.emit(self.n) time.sleep(self.n*1.1) self.signals.completed.emit(self.n)
Code language: Python (python)

First, initialize the job number (n) and Signals object in the __init__() method.

Second, override the run() method of the QRunnable class. To simulate a long-running task, we use the sleep() function of the time module. Before starting the timer, we emit the started signal; after the timer completes, we emit the completed signal.

MainWindow class

The MainWindow class defines the UI for the application:

class MainWindow(QMainWindow): def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle('QThreadPool Demo') self.comleted_jobs = [] self.job_count = 10 widget = QWidget() widget.setLayout(QGridLayout()) self.setCentralWidget(widget) self.btn_start = QPushButton('Start', clicked=self.start_jobs) self.progress_bar = QProgressBar(minimum=0, maximum=self.job_count) self.list = QListWidget() widget.layout().addWidget(self.list, 0, 0, 1, 2) widget.layout().addWidget(self.progress_bar, 1, 0) widget.layout().addWidget(self.btn_start, 1, 1) self.show() def start_jobs(self): self.restart() pool = QThreadPool.globalInstance() for i in range(1, self.job_count+1): runnable = Worker(i) runnable.signals.completed.connect(self.complete) runnable.signals.started.connect(self.start) pool.start(runnable) def restart(self): self.progress_bar.setValue(0) self.comleted_jobs = [] self.btn_start.setEnabled(False) def start(self, n): self.list.addItem(f'Job #{n} started...') def complete(self, n): self.list.addItem(f'Job #{n} completed.') self.comleted_jobs.append(n) self.progress_bar.setValue(len(self.comleted_jobs)) if len(self.comleted_jobs) == self.job_count: self.btn_start.setEnabled(True)
Code language: Python (python)

First, initialize the number of jobs (job_count) and completed_jobs list in the __init__() method of the MainWindow class:

self.job_count = 10 self.comleted_jobs = []
Code language: Python (python)

Second, define the start_jobs() method that will be executed when the user clicks the start button:

def start_jobs(self): self.restart() pool = QThreadPool.globalInstance() for i in range(1, self.job_count+1): worker = Worker(i) worker.signals.completed.connect(self.complete) worker.signals.started.connect(self.start) pool.start(worker)
Code language: Python (python)

The restart() resets the completed_jobs, updates the progress bar to zero, and disables the start button:

def restart(self): self.progress_bar.setValue(0) self.comleted_jobs = [] self.btn_start.setEnabled(False)
Code language: Python (python)

To get the QThreadPool object, we use the globalInstance() of the QThreadPool class:

pool = QThreadPool.globalInstance()
Code language: Python (python)

We create a number of workers, connect their signals to the methods of the MainWindow class, and start worker threads using the start() method of the QThreadPool object.

The start() method adds the message that starts a worker thread to the QListWidget:

def start(self, n): self.list.addItem(f'Job #{n} started...')
Code language: Python (python)

The completed() method runs each time a worker thread is completed. It adds a message to the QListWidget, updates the progress bar, and enables the start button if all worker threads are completed:

def complete(self, n): self.list.addItem(f'Job #{n} completed.') self.comleted_jobs.append(n) self.progress_bar.setValue(len(self.comleted_jobs)) if len(self.comleted_jobs) == self.job_count: self.btn_start.setEnabled(True)
Code language: Python (python)

Using QThreadPool to get stock prices

The following Stock Listing program reads stock symbols from symbols.txt file and uses QThreadPool to get the stock prices from Yahoo Finance website:

Stock Listing Program:

import sys from pathlib import Path from PyQt6.QtCore import QRunnable, Qt, QObject, QThreadPool, pyqtSignal as Signal, pyqtSlot as Slot from PyQt6.QtWidgets import QApplication, QMainWindow, QPushButton, QWidget, QGridLayout, QProgressBar, QTableWidget, QTableWidgetItem from PyQt6.QtGui import QIcon from lxml import html import requests class Signals(QObject): completed = Signal(dict) class Stock(QRunnable): BASE_URL = 'https://finance.yahoo.com/quote/' def __init__(self, symbol): super().__init__() self.symbol = symbol self.signal = Signals() @Slot() def run(self): stock_url = f'{self.BASE_URL}{self.symbol}' response = requests.get(stock_url) if response.status_code != 200: self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'}) return tree = html.fromstring(response.text) price_text = tree.xpath( '//*[@id="quote-header-info"]/div[3]/div[1]/div[1]/fin-streamer[1]/text()' ) if not price_text: self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'}) return price = float(price_text[0].replace(',', '')) self.signal.completed.emit({'symbol': self.symbol, 'price': price}) class Window(QMainWindow): def __init__(self, filename, *args, **kwargs): super().__init__(*args, **kwargs) self.symbols = self.read_symbols(filename) self.results = [] self.setWindowTitle('Stock Listing') self.setGeometry(100, 100, 400, 300) self.setWindowIcon(QIcon('./assets/stock.png')) widget = QWidget() widget.setLayout(QGridLayout()) self.setCentralWidget(widget) # set up button & progress bar self.btn_start = QPushButton('Get Prices', clicked=self.get_prices) self.progress_bar = QProgressBar(minimum=1, maximum=len(self.symbols)) # set up table widget self.table = QTableWidget(widget) self.table.setColumnCount(2) self.table.setColumnWidth(0, 150) self.table.setColumnWidth(1, 150) self.table.setHorizontalHeaderLabels(['Symbol', 'Price']) widget.layout().addWidget(self.table, 0, 0, 1, 2) widget.layout().addWidget(self.progress_bar, 1, 0) widget.layout().addWidget(self.btn_start, 1, 1) # show the window self.show() def read_symbols(self, filename): """ Read symbols from a file """ path = Path(filename) text = path.read_text() return [symbol.strip() for symbol in text.split('\n')] def reset_ui(self): self.progress_bar.setValue(1) self.table.setRowCount(0) def get_prices(self): # reset ui self.reset_ui() # start worker threads pool = QThreadPool.globalInstance() stocks = [Stock(symbol) for symbol in self.symbols] for stock in stocks: stock.signal.completed.connect(self.update) pool.start(stock) def update(self, data): # add a row to the table row = self.table.rowCount() self.table.insertRow(row) self.table.setItem(row, 0, QTableWidgetItem(data['symbol'])) self.table.setItem(row, 1, QTableWidgetItem(str(data['price']))) # update the progress bar self.progress_bar.setValue(row + 1) # sort the list by symbols once completed if row == len(self.symbols) - 1: self.table.sortItems(0, Qt.SortOrder.AscendingOrder) if __name__ == '__main__': app = QApplication(sys.argv) window = Window('symbols.txt') sys.exit(app.exec())
Code language: Python (python)

How it works.

Signals class

We define the Signals class that is a subclass of the QObject. The Signals class has one class variable completed which is an instance of the Signal class.

The completed signal holds a dictionary and is emitted once the program completes getting the stock price.

class Signals(QObject): completed = Signal(dict)
Code language: Python (python)

Stock class

The STock class inherits from the QRunnable class. It overrides the run() method that gets the stock price from the Yahoo Finance website.

Once completed, the run() method emits the completed signal with the stock symbol and price.

If an error occurs like the symbol is not found or the website changes the way it displays the stock price, the run() method returns the symbol with the price as a N/A string.

class Stock(QRunnable): BASE_URL = 'https://finance.yahoo.com/quote/' def __init__(self, symbol): super().__init__() self.symbol = symbol self.signal = Signals() @Slot() def run(self): stock_url = f'{self.BASE_URL}{self.symbol}' response = requests.get(stock_url) if response.status_code != 200: self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'}) return tree = html.fromstring(response.text) price_text = tree.xpath( '//*[@id="quote-header-info"]/div[3]/div[1]/div[1]/fin-streamer[1]/text()' ) if not price_text: self.signal.completed.emit({'symbol': self.symbol, 'price': 'N/A'}) return price = float(price_text[0].replace(',', '')) self.signal.completed.emit({'symbol': self.symbol, 'price': price})
Code language: Python (python)

Note that Yahoo Finance may change its structure. To make the program works, you need to change the XPath of the price of the new one:

//*[@id="quote-header-info"]/div[3]/div[1]/div[1]/fin-streamer[1]/text()
Code language: Python (python)

MainWindow Class

First, read the symbols from a file and assign them to the self.symbols variables:

self.symbols = self.read_symbols(filename)
Code language: Python (python)

The read_symbols() method looks like this:

def read_symbols(self, filename): path = Path(filename) text = path.read_text() return [symbol.strip() for symbol in text.split('\n')]
Code language: Python (python)

The text file (symbols.txt) contains each symbol per line:

AAPL MSFT GOOG AMZN TSLA META NVDA BABA CRM INTC PYPL AMD ATVI EA TTD ORCL
Code language: Python (python)

Second, define the get_prices that uses the QThreadPool to create worker threads for getting stock prices:

def get_prices(self): # reset ui self.reset_ui() # start worker threads pool = QThreadPool.globalInstance() stocks = [Stock(symbol) for symbol in self.symbols] for stock in stocks: stock.signal.completed.connect(self.update) pool.start(stock)
Code language: Python (python)

The reset_ui() method clear all rows of the QTableWidget and set the progress bar to its minimum value:

def reset_ui(self): self.table.setRowCount(0) self.progress_bar.setValue(1)
Code language: Python (python)

Third, define the update() method that will be called once each worker thread is completed. The update() method adds a new row to the table, updates the progress bar, and sorts the symbols once all the worker threads are completed:

def update(self, data): # add a row to the table row = self.table.rowCount() self.table.insertRow(row) self.table.setItem(row, 0, QTableWidgetItem(data['symbol'])) self.table.setItem(row, 1, QTableWidgetItem(str(data['price']))) # update the progress bar self.progress_bar.setValue(row + 1) # sort the list by symbols once completed if row == len(self.symbols) - 1: self.table.sortItems(0, Qt.SortOrder.AscendingOrder)
Code language: Python (python)

Summary

  • Use the QRunnable class to represent a long-running task that will be offloaded to a worker thread.
  • Use the QThreadPool class to manage worker threads automatically.
  • Each PyQt application has one QThreadPool object. Use the globalInstance() method to get the global QThreadPool object.
  • Use the start() method of the QThreadPool object to start a worker thread.
Did you find this tutorial helpful ?