How to Run Custom Functions with Arguments in QThreadPool and QRunnable

A common challenge in PyQt6 applications is running long calculations without freezing the GUI. You might have several custom functions — each performing different calculations — and you want to call them with specific arguments while keeping your interface responsive. Let's look at how to do this properly.

Table of contents

The problem
Using QThreadPool to run functions in the background
Passing arguments to your functions
A complete working example
Passing keyword arguments
A note about thread safety
When to use QProcess instead
Summary

The problem

Imagine you have a PyQt6 application with a button that triggers a calculation. Depending on user selections from combo boxes and text fields, you call different functions with different arguments. Something like this:

python

def calc(self):
    indicator = self.combo_indicator.currentText()
    output = self.combo_output.currentText()
    version = self.txt_version.text()
    year = self.txt_year.text()

    # This blocks the GUI until it finishes!
    result = some_long_calculation(indicator, output, version, year)

While some_long_calculation runs, your entire GUI freezes. Buttons stop responding, windows can't be moved, and your operating system might even flag the application as "not responding." That's because the calculation is running on the main thread — the same thread responsible for drawing the interface and handling user input.

For running your own Python functions in the background, threads are the way to go. PyQt6 provides QThreadPool and QRunnable for exactly this purpose. For a comprehensive introduction to this topic, see the full Multithreading PyQt6 applications with QThreadPool tutorial.

Using QThreadPool to run functions in the background

PyQt6's QThreadPool manages a pool of worker threads. You can submit tasks to it, and it will run them on available threads without blocking your GUI. Combined with a custom QRunnable wrapper, you can pass any function — along with its arguments — to run in the background.

Here's a reusable Worker class that wraps any function for use with QThreadPool:

python

import traceback
import sys
from PyQt6.QtCore import QRunnable, pyqtSlot, pyqtSignal, QObject


class WorkerSignals(QObject):
    """
    Signals available from a running worker thread.

    finished
        No data

    error
        tuple (exctype, value, traceback.format_exc())

    result
        object data returned from the function

    """
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)


class Worker(QRunnable):
    """
    Worker thread

    Inherits from QRunnable to handle worker thread setup, signals and
    wrap-up.

    :param fn: The function to run on this worker thread. Supplied args
               and kwargs will be passed through to the runner.
    :param args: Arguments to pass to the callback function
    :param kwargs: Keywords to pass to the callback function

    """

    def __init__(self, fn, *args, **kwargs):
        super().__init__()
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

    @pyqtSlot()
    def run(self):
        """
        Run the function with passed args, kwargs.
        """
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            self.signals.result.emit(result)
        finally:
            self.signals.finished.emit()

The Worker class accepts a function (fn) plus any positional and keyword arguments. When the thread pool runs the worker, it calls that function with those arguments. The WorkerSignals class provides signals to notify your GUI when the function finishes, returns a result, or encounters an error.

Why do we need a separate WorkerSignals class? In Qt, signals must be defined on QObject subclasses. Since QRunnable doesn't inherit from QObject, we create a small QObject-based class to hold the signals and attach an instance of it to each worker.

Passing arguments to your functions

Now let's use this Worker class to run custom functions with arguments. Suppose you have a couple of calculation functions like these:

python

def calculate_indicator(indicator, output, version, year):
    """A long-running calculation that returns a result."""
    import time
    time.sleep(5)  # Simulating heavy work
    return f"Result for {indicator}, {output}, {version}, {year}"


def calculate_summary(indicator, year):
    """Another long-running calculation."""
    import time
    time.sleep(3)
    return f"Summary for {indicator} in {year}"

In your main window, you'd set up the thread pool and wire everything together:

python

from PyQt6.QtWidgets import QMainWindow
from PyQt6.QtCore import QThreadPool


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.threadpool = QThreadPool()
        print(
            f"Multithreading with maximum "
            f"{self.threadpool.maxThreadCount()} threads"
        )
        # ... set up your UI widgets here ...

When the user clicks the Calculate button, you create a Worker with the appropriate function and arguments, connect the signals, and hand it to the thread pool:

python

def calc(self):
    indicator = self.combo_indicator.currentText()
    output = self.combo_output.currentText()
    version = self.txt_version.text()
    year = self.txt_year.text()

    # Create a worker with the function and its arguments
    worker = Worker(
        calculate_indicator, indicator, output, version, year
    )

    # Connect signals
    worker.signals.result.connect(self.handle_result)
    worker.signals.error.connect(self.handle_error)
    worker.signals.finished.connect(self.handle_finished)

    # Disable the button while calculating
    self.btn_calculate.setEnabled(False)
    self.label_status.setText("Calculating...")

    # Submit to the thread pool
    self.threadpool.start(worker)

The pattern is always the same: create a Worker with your function and arguments, connect the signals you care about, then start it on the thread pool.

The finished signal re-enables the Calculate button once the work is done. The error signal catches any exceptions so your app doesn't silently fail. And the result signal delivers whatever your function returns back to the GUI thread, where it's safe to update widgets.

python

def handle_result(self, result):
    self.label_status.setText(str(result))

def handle_error(self, error):
    exctype, value, tb_str = error
    self.label_status.setText(f"Error: {value}")
    print(tb_str)

def handle_finished(self):
    self.btn_calculate.setEnabled(True)

You can use this same approach for any function. Want to run calculate_summary instead? Just swap the function and arguments:

python

worker = Worker(calculate_summary, indicator, year)

Everything else — the signal connections, the thread pool management — stays exactly the same.

A complete working example

Here's a full, self-contained example you can copy and run. It simulates long-running calculations using time.sleep and shows how different functions receive their arguments on a background thread:

python

import sys
import time
import traceback

from PyQt6.QtCore import (
    QObject,
    QRunnable,
    QThreadPool,
    pyqtSignal,
    pyqtSlot,
)
from PyQt6.QtWidgets import (
    QApplication,
    QComboBox,
    QHBoxLayout,
    QLabel,
    QLineEdit,
    QMainWindow,
    QPushButton,
    QVBoxLayout,
    QWidget,
)


# ---- Worker classes --------------------------------------------------------


class WorkerSignals(QObject):
    """
    Signals available from a running worker thread.
    """
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)


class Worker(QRunnable):
    """
    Worker thread – wraps any callable with arguments for use
    with QThreadPool.
    """

    def __init__(self, fn, *args, **kwargs):
        super().__init__()
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.signals = WorkerSignals()

    @pyqtSlot()
    def run(self):
        try:
            result = self.fn(*self.args, **self.kwargs)
        except Exception:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit(
                (exctype, value, traceback.format_exc())
            )
        else:
            self.signals.result.emit(result)
        finally:
            self.signals.finished.emit()


# ---- Example calculation functions -----------------------------------------


def calculate_indicator(indicator, output, version, year):
    """Simulate a long-running indicator calculation."""
    print(
        f"[Thread] Running calculate_indicator("
        f"{indicator}, {output}, {version}, {year})"
    )
    time.sleep(5)  # Simulate heavy work
    return (
        f"Indicator result: {indicator} / {output} "
        f"(version {version}, year {year})"
    )


def calculate_summary(indicator, year):
    """Simulate a different long-running calculation."""
    print(
        f"[Thread] Running calculate_summary("
        f"{indicator}, {year})"
    )
    time.sleep(3)
    return f"Summary result: {indicator} for {year}"


# ---- Main window -----------------------------------------------------------


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("QThreadPool Demo")

        # Thread pool
        self.threadpool = QThreadPool()
        print(
            f"Multithreading with maximum "
            f"{self.threadpool.maxThreadCount()} threads"
        )

        # --- Widgets ---
        self.combo_calc_type = QComboBox()
        self.combo_calc_type.addItems(["Indicator", "Summary"])

        self.combo_indicator = QComboBox()
        self.combo_indicator.addItems(["IND_001", "IND_002", "IND_003"])

        self.combo_output = QComboBox()
        self.combo_output.addItems(["OTD_001", "OTD_002"])

        self.txt_version = QLineEdit("v1.0")
        self.txt_year = QLineEdit("2024")

        self.btn_calculate = QPushButton("Calculate")
        self.btn_calculate.clicked.connect(self.calc)

        self.label_status = QLabel("Ready")

        # --- Layout ---
        form_layout = QHBoxLayout()
        form_layout.addWidget(QLabel("Type:"))
        form_layout.addWidget(self.combo_calc_type)
        form_layout.addWidget(QLabel("Indicator:"))
        form_layout.addWidget(self.combo_indicator)
        form_layout.addWidget(QLabel("Output:"))
        form_layout.addWidget(self.combo_output)
        form_layout.addWidget(QLabel("Version:"))
        form_layout.addWidget(self.txt_version)
        form_layout.addWidget(QLabel("Year:"))
        form_layout.addWidget(self.txt_year)

        main_layout = QVBoxLayout()
        main_layout.addLayout(form_layout)
        main_layout.addWidget(self.btn_calculate)
        main_layout.addWidget(self.label_status)

        container = QWidget()
        container.setLayout(main_layout)
        self.setCentralWidget(container)

    def calc(self):
        """
        Read values from the UI, pick the right function, and
        run it on a background thread.
        """
        calc_type = self.combo_calc_type.currentText()
        indicator = self.combo_indicator.currentText()
        output = self.combo_output.currentText()
        version = self.txt_version.text()
        year = self.txt_year.text()

        # Choose the function and arguments based on user selection
        if calc_type == "Indicator":
            worker = Worker(
                calculate_indicator, indicator, output, version, year
            )
        else:
            worker = Worker(
                calculate_summary, indicator, year
            )

        # Connect signals to handler methods
        worker.signals.result.connect(self.handle_result)
        worker.signals.error.connect(self.handle_error)
        worker.signals.finished.connect(self.handle_finished)

        # Update UI and start the worker
        self.btn_calculate.setEnabled(False)
        self.label_status.setText("Calculating...")
        self.threadpool.start(worker)

    def handle_result(self, result):
        """Display the result returned by the worker function."""
        self.label_status.setText(str(result))

    def handle_error(self, error):
        """Display error information if the worker function failed."""
        exctype, value, tb_str = error
        self.label_status.setText(f"Error: {value}")
        print(tb_str)

    def handle_finished(self):
        """Re-enable the button when the worker is done."""
        self.btn_calculate.setEnabled(True)


# ---- Run ------------------------------------------------------------------

app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec())

Run this example and click Calculate. You'll see the status label update to "Calculating..." while the function runs in the background. The GUI stays responsive — you can still interact with the combo boxes and text fields. When the calculation finishes, the result appears in the status label and the button re-enables.

Try switching the calculation type combo box between "Indicator" and "Summary" to see how different functions with different argument signatures are dispatched through the same Worker pattern.

Passing keyword arguments

The Worker class supports keyword arguments too. This is useful when your functions have optional parameters:

python

def calculate_indicator(indicator, output, version, year, verbose=False):
    if verbose:
        print(f"Starting calculation for {indicator}...")
    time.sleep(5)
    return f"Result: {indicator} / {output}"

Pass keyword arguments just as you normally would when creating the worker:

python

worker = Worker(
    calculate_indicator,
    indicator, output, version, year,
    verbose=True
)

The Worker.__init__ method captures *args and **kwargs, then unpacks them in run() when calling the function. Your function receives everything exactly as if you'd called it directly.

A note about thread safety

When your function runs on a background thread, it shouldn't directly modify any Qt widgets. Widget updates must happen on the main thread. That's exactly what the signals are for — when you emit result with your return value, the connected slot runs on the main thread where it's safe to update labels, tables, progress bars, or anything else in your UI.

If your background function needs to report progress during execution, you can add a progress signal to WorkerSignals and emit it from within your function. To do this, pass the signals object into your function as an argument:

python

class WorkerSignals(QObject):
    finished = pyqtSignal()
    error = pyqtSignal(tuple)
    result = pyqtSignal(object)
    progress = pyqtSignal(int)


def long_calculation(data, signals):
    """A function that reports progress as it works."""
    total = len(data)
    results = []
    for i, item in enumerate(data):
        time.sleep(0.5)  # Simulate processing each item
        results.append(item * 2)
        signals.progress.emit(int((i + 1) / total * 100))
    return results

Then when creating the worker, pass the signals object as an argument:

python

worker = Worker(long_calculation, my_data, worker.signals)
worker.signals.progress.connect(self.update_progress_bar)

Since signals are thread-safe in Qt, emitting them from a background thread is perfectly fine. The connected slot will execute on the main thread.

When to use QProcess instead

If your functions are defined in separate Python scripts and you genuinely want to run them as independent processes, QProcess is the right tool. In that case, you'd structure things differently — saving your function logic to a standalone script and calling it with QProcess:

python

from PyQt6.QtCore import QProcess


def start_external_calculation(self):
    self.process = QProcess()
    self.process.finished.connect(self.process_finished)
    self.process.start(
        "python",
        ["my_calculation_script.py", "IND_001", "OTD_002", "v1.0", "2024"],
    )

The arguments are passed as a list of strings, similar to how you'd run a command on the terminal. Your external script would then read them from sys.argv.

This approach gives you true process isolation — if the calculation crashes, your GUI keeps running — but it adds complexity around communication between processes. For most cases where you're calling Python functions within the same application, QThreadPool with the Worker pattern shown above is simpler and more practical.

Summary

When you want to run user-defined functions with arguments in the background:

Use QThreadPool and QRunnable for Python functions that live inside your application. Pass the function and its arguments to a Worker class, and use signals to communicate results back to the GUI. For more details on using @pyqtSlot decorators as shown in the Worker class, see What does @pyqtSlot() do?
Use QProcess when you need to run a completely separate script or executable. Arguments are passed as command-line strings.

The Worker pattern gives you a clean, reusable way to offload any function to a background thread — just pass the function and its arguments, connect the signals, and let the thread pool handle the rest.

Well done, you've finished this tutorial! Mark As Complete

[[ user.completed.length ]] completed [[ user.streak+1 ]] day streak