When you're working with multiple WorkerManager instances — each with their own thread pools and worker queues — you might want to display progress for all of them in a single QListView. The problem is that a view in Qt can only have one model set on it at a time. So how do you get workers from multiple managers to show up in the same list?
The answer is to separate the model (which holds the state) from the manager (which runs the workers). You create a single shared model, pass it to each manager, and point the view at that model. All managers update the same model, and the view displays everything in one place.
The problem: one view, one model
In Qt's Model/View architecture, a view like QListView pulls its data from a single model. You set it with .setModel():
self.progress = QListView()
self.progress.setModel(some_model)
There's no .addModel() or way to combine models directly on the view. If you call .setModel() a second time, it replaces the first model entirely.
So the trick is to make sure all the data you want displayed ends up in one model.
The solution: a shared model
Instead of having each WorkerManager act as both a manager and a model, we split things up:
WorkerModel— aQAbstractListModelsubclass that holds the state (progress, status) for all workers.WorkerManager— aQObjectsubclass that manages the thread pool and worker queue, but delegates state tracking to a sharedWorkerModel.
Each WorkerManager receives a reference to the same WorkerModel instance. When a worker reports progress or changes status, the manager forwards that update to the shared model. The QListView is connected to this single model, so it sees updates from all managers.
Here's how the pieces fit together in MainWindow:
self.model = WorkerModel()
self.workers1 = WorkerManager(self.model)
self.workers2 = WorkerManager(self.model)
self.progress = QListView()
self.progress.setModel(self.model)
Both workers1 and workers2 write to the same self.model. The QListView reads from that model. Done.
The WorkerModel
The WorkerModel is a QAbstractListModel that stores a dictionary of worker states, keyed by job ID. Each entry tracks the worker's progress and status.
class WorkerModel(QAbstractListModel):
_state = {}
def add(self, identifier, data):
self._state[identifier] = data
self.layoutChanged.emit()
def data(self, index, role):
if role == Qt.DisplayRole:
job_ids = list(self._state.keys())
job_id = job_ids[index.row()]
return job_id, self._state[job_id]
def rowCount(self, index):
return len(self._state)
def receive_status(self, job_id, status):
self._state[job_id]["status"] = status
self.layoutChanged.emit()
def receive_progress(self, job_id, progress):
self._state[job_id]["progress"] = progress
self.layoutChanged.emit()
def cleanup(self):
for job_id, s in list(self._state.items()):
if s["status"] in (STATUS_COMPLETE, STATUS_ERROR):
del self._state[job_id]
self.layoutChanged.emit()
The receive_status and receive_progress methods are slots that connect directly to worker signals. When any worker — regardless of which manager started it — emits a progress or status update, it lands here and triggers a view refresh via layoutChanged.
The WorkerManager
The WorkerManager no longer subclasses QAbstractListModel. It's a plain QObject that holds the thread pool and connects worker signals to the shared model:
class WorkerManager(QObject):
_workers = {}
status = pyqtSignal(str)
def __init__(self, model):
super().__init__()
self.model = model
self.threadpool = QThreadPool()
self.max_threads = self.threadpool.maxThreadCount()
def enqueue(self, worker):
worker.signals.status.connect(self.model.receive_status)
worker.signals.progress.connect(self.model.receive_progress)
worker.signals.finished.connect(self.done)
self.threadpool.start(worker)
self._workers[worker.job_id] = worker
self.model.add(worker.job_id, DEFAULT_STATE.copy())
When enqueue is called, the manager wires the worker's signals to the model's slots, not its own. The model gets updated directly, and the view refreshes automatically. For more on using QThreadPool to manage concurrent workers, see Multithreading PyQt6 applications with QThreadPool.
Complete working example
Here's the full application. You can copy and run this directly. Click "Start a worker 1" or "Start a worker 2" to enqueue workers from different managers — they all appear in the same list view.
import random
import sys
import time
import uuid
from PyQt6.QtCore import (
QAbstractListModel,
QObject,
QRect,
QRunnable,
Qt,
QThreadPool,
QTimer,
pyqtSignal,
pyqtSlot,
)
from PyQt6.QtGui import QBrush, QColor, QPen
from PyQt6.QtWidgets import (
QApplication,
QListView,
QMainWindow,
QPlainTextEdit,
QPushButton,
QStyledItemDelegate,
QVBoxLayout,
QWidget,
)
STATUS_WAITING = "waiting"
STATUS_RUNNING = "running"
STATUS_ERROR = "error"
STATUS_COMPLETE = "complete"
STATUS_COLORS = {
STATUS_RUNNING: "#33a02c",
STATUS_ERROR: "#e31a1c",
STATUS_COMPLETE: "#b2df8a",
}
DEFAULT_STATE = {"progress": 0, "status": STATUS_WAITING}
class WorkerSignals(QObject):
"""
Signals available from a running worker thread.
"""
error = pyqtSignal(str, str)
result = pyqtSignal(str, object)
finished = pyqtSignal(str)
progress = pyqtSignal(str, int)
status = pyqtSignal(str, str)
class Worker(QRunnable):
"""
Worker thread that performs a simple calculation with
random delays to simulate real work.
"""
def __init__(self, *args, **kwargs):
super().__init__()
self.signals = WorkerSignals()
self.job_id = str(uuid.uuid4())
self.args = args
self.kwargs = kwargs
self.signals.status.emit(self.job_id, STATUS_WAITING)
@pyqtSlot()
def run(self):
self.signals.status.emit(self.job_id, STATUS_RUNNING)
x, y = self.args
try:
value = random.randint(0, 100) * x
delay = random.random() / 10
result = []
for n in range(100):
value = value / y
y -= 1
result.append(value)
self.signals.progress.emit(self.job_id, n + 1)
time.sleep(delay)
except Exception as e:
print(e)
self.signals.error.emit(self.job_id, str(e))
self.signals.status.emit(self.job_id, STATUS_ERROR)
else:
self.signals.result.emit(self.job_id, result)
self.signals.status.emit(self.job_id, STATUS_COMPLETE)
self.signals.finished.emit(self.job_id)
class WorkerModel(QAbstractListModel):
"""
Shared model that holds worker state (progress, status)
for display in a view.
"""
def __init__(self):
super().__init__()
self._state = {}
def add(self, identifier, data):
self._state[identifier] = data
self.layoutChanged.emit()
def data(self, index, role):
if role == Qt.DisplayRole:
job_ids = list(self._state.keys())
job_id = job_ids[index.row()]
return job_id, self._state[job_id]
def rowCount(self, index):
return len(self._state)
def receive_status(self, job_id, status):
self._state[job_id]["status"] = status
self.layoutChanged.emit()
def receive_progress(self, job_id, progress):
self._state[job_id]["progress"] = progress
self.layoutChanged.emit()
def cleanup(self):
"""
Remove any complete or failed workers from the state.
"""
for job_id, s in list(self._state.items()):
if s["status"] in (STATUS_COMPLETE, STATUS_ERROR):
del self._state[job_id]
self.layoutChanged.emit()
class WorkerManager(QObject):
"""
Manages a thread pool and worker queue. Delegates state
tracking to a shared WorkerModel.
"""
status = pyqtSignal(str)
def __init__(self, model):
super().__init__()
self.model = model
self._workers = {}
self.threadpool = QThreadPool()
self.max_threads = self.threadpool.maxThreadCount()
print(
"Multithreading with maximum %d threads" % self.max_threads
)
self.status_timer = QTimer()
self.status_timer.setInterval(100)
self.status_timer.timeout.connect(self.notify_status)
self.status_timer.start()
def notify_status(self):
n_workers = len(self._workers)
running = min(n_workers, self.max_threads)
waiting = max(0, n_workers - self.max_threads)
self.status.emit(
"{} running, {} waiting, {} threads".format(
running, waiting, self.max_threads
)
)
def enqueue(self, worker):
"""
Enqueue a worker to run by passing it to the QThreadPool.
"""
worker.signals.error.connect(self.receive_error)
worker.signals.status.connect(self.model.receive_status)
worker.signals.progress.connect(self.model.receive_progress)
worker.signals.finished.connect(self.done)
self.threadpool.start(worker)
self._workers[worker.job_id] = worker
# Set default state: waiting, 0 progress.
self.model.add(worker.job_id, DEFAULT_STATE.copy())
def receive_error(self, job_id, message):
print(job_id, message)
def done(self, job_id):
"""
Worker complete. Remove from active workers dict.
We leave it in the model so completed workers are
still visible.
"""
del self._workers[job_id]
class ProgressBarDelegate(QStyledItemDelegate):
def paint(self, painter, option, index):
job_id, data = index.model().data(index, Qt.DisplayRole)
if data["progress"] > 0:
color = QColor(STATUS_COLORS[data["status"]])
brush = QBrush()
brush.setColor(color)
brush.setStyle(Qt.SolidPattern)
width = option.rect.width() * data["progress"] / 100
rect = QRect(option.rect)
rect.setWidth(int(width))
painter.fillRect(rect, brush)
pen = QPen()
pen.setColor(Qt.black)
painter.drawText(option.rect, Qt.AlignLeft, job_id)
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.model = WorkerModel()
self.workers1 = WorkerManager(self.model)
self.workers2 = WorkerManager(self.model)
self.workers1.status.connect(self.statusBar().showMessage)
self.workers2.status.connect(self.statusBar().showMessage)
layout = QVBoxLayout()
self.progress = QListView()
self.progress.setModel(self.model)
delegate = ProgressBarDelegate()
self.progress.setItemDelegate(delegate)
layout.addWidget(self.progress)
self.text = QPlainTextEdit()
self.text.setReadOnly(True)
start1 = QPushButton("Start a worker 1")
start1.pressed.connect(self.start_worker1)
start2 = QPushButton("Start a worker 2")
start2.pressed.connect(self.start_worker2)
clear = QPushButton("Clear")
clear.pressed.connect(self.model.cleanup)
layout.addWidget(self.text)
layout.addWidget(start1)
layout.addWidget(start2)
layout.addWidget(clear)
w = QWidget()
w.setLayout(layout)
self.setCentralWidget(w)
self.show()
def start_worker1(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers1.enqueue(w)
def start_worker2(self):
x = random.randint(0, 1000)
y = random.randint(0, 1000)
w = Worker(x, y)
w.signals.result.connect(self.display_result)
w.signals.error.connect(self.display_result)
self.workers2.enqueue(w)
def display_result(self, job_id, data):
self.text.appendPlainText("WORKER %s: %s" % (job_id, data))
app = QApplication(sys.argv)
window = MainWindow()
app.exec()
When you run this, you'll see a window with two buttons — one for each worker manager. Clicking either button starts a new worker, and all workers appear together in the same QListView, complete with progress bars. The "Clear" button removes any finished or errored workers from the list.
This pattern — creating a shared model and injecting it into multiple managers — works well whenever you need to aggregate data from several sources into a single view. Each manager stays independent (with its own thread pool), but they all contribute to one unified display. If you're looking for more examples of using custom delegates and table models with views, the QTableView with numpy and pandas tutorial covers related concepts for tabular data.
Create GUI Applications with Python & Qt6 by Martin Fitzpatrick
(PyQt6 Edition) The hands-on guide to making apps with Python — Over 15,000 copies sold!
