Sharing class variables across thread (QgsTasks) with pyQGIS

Question

In keeping with Setting up thread with busy progress indicator or GIF in PyQGIS where I asked about thread and busy indicator I now wish to ask understand how it is possible to use a variable declared in an instance of a Qgstask in another Qgstask?

In the code below I tried to use the output of the native 'centroid' QGIS tool as the input of the buffering tool used in another Qgstask. I've done a lot of researches on internet but I don't find any examples about my problem.

from qgis.PyQt.QtCore import QSettings, QTranslator, QCoreApplication
from qgis.PyQt.QtGui import *
from qgis.PyQt.QtWidgets import *

from datetime import datetime
import os.path, os, shutil
from qgis.core import *
from qgis.gui import *
from qgis.utils import *
import processing
from PyQt5.QtCore import pyqtSignal


class MyTask(QgsTask):
    messaging = pyqtSignal(str)
    result = pyqtSignal(dict)

    def __init__(self, layer):
        super(MyTask, self).__init__()
        self.layer = layer

    def run(self):
        start = datetime.now()
        # buffer
        alg_params = {
            'ALL_PARTS': True,
            'INPUT': self.layer,
            'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
        }
        self.centroid = processing.run('native:centroids', alg_params)

        end = datetime.now()

        buffer_info = "Making centroids..."
        buffer_title = '<center><h3><span style=\" color: #7eb346; font-weight:bold\">%s</span></h3></center>' % buffer_info
        self.messaging.emit(buffer_title)
        self.messaging.emit(
            "Time required to achieve the 1st task: " + str((end - start).total_seconds()) + " seconds" + "\n")

        return True


class MyTask2(QgsTask):

    messaging = pyqtSignal(str)
    result = pyqtSignal(dict)

    def __init__(self, centroid):
        super(MyTask2, self).__init__()
        self.centroid = centroid

    def run(self):
        start2 = datetime.now()
        # buffer
        alg_params = {
            'DISSOLVE': True,
            'DISTANCE': 100,
            'END_CAP_STYLE': 0,
            'INPUT': self.centroid['OUTPUT'],
            'JOIN_STYLE': 0,
            'MITER_LIMIT': 2,
            'SEGMENTS': 5,
            'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
        }

        buffer = processing.runAndLoadResults('native:buffer', alg_params)

        end2 = datetime.now()

        buffer_info = "Layer Buffering..."
        buffer_title = '<center><h3><span style=\" color: #7eb346; font-weight:bold\">%s</span></h3></center>' % buffer_info
        self.messaging.emit(buffer_title)
        self.messaging.emit("Time required to achieve the task: " + str((end2 - start2).total_seconds()) + " seconds" + "\n")

        return True


class plugintestDialog(QDialog):
    def __init__(self, parent=None):
        super(plugintestDialog, self).__init__(parent)
        self.lyt = QVBoxLayout(self)
        self.pushButton = QPushButton("Test 1", self)
        self.progressbar = QProgressBar(self)
        self.progressbar.setMinimum(0)
        self.lyt.addWidget(self.pushButton)
        self.lyt.addWidget(self.progressbar)


class plugintest:

    messaging = pyqtSignal(str)

    def __init__(self, iface):
        self.iface = iface
        self.dlg = plugintestDialog()
        self.dlg.pushButton.clicked.connect(self.centroid)

    def run(self):
        self.dlg.show()
        result = self.dlg.exec_()

    def centroid(self):
        # disable pushbutton for avoid multi-clicks
        self.dlg.pushButton.setEnabled(False)
        # start the busy state of the progressbar
        self.dlg.progressbar.setMaximum(0)
        # define the layer to process (just comment / uncomment what you want)
        self.layer = QgsProject.instance().mapLayersByName('layer')[0]
        # self.layer  = self.dlg.mMapLayerComboBox.currentLayer()
        # create a new task
        self.buffer_task = MyTask(self.layer)
        # connect the signals
        self.centroid_task.messaging.connect(lambda t: print(t))
        self.centroid_task.taskCompleted.connect(self.buffering)
        # launch the processing of the task
        QgsApplication.taskManager().addTask(self.centroid_task)

    def buffering(self):
        self.centroid= self.centroid['OUTPUT']
        self.buffer_task = MyTask2(self.centroid)
        # connect the signals
        self.buffer_task.messaging.connect(lambda t: print(t))
        self.buffer_task.taskCompleted.connect(self.buffering_completed)
        # launch the processing of the task
        QgsApplication.taskManager().addTask(self.buffer_task)

    def buffering_completed(self):
        self.dlg.progressbar.setMaximum(1)
        self.buffer_task.disconnect()
        self.dlg.pushButton.setEnabled(True)


my_instance = plugintest(iface)
my_instance.run()

Answer 1

Here is working example. I have modified your code a bit. Because both your tasks have much the same function (to run a processing algorithm and return a result layer) I would prefer to only create one QgsTask subclass and pass all necessary parameters as variables to the subclass constructor. As @J. Monticolo already told you, you connect to the result signal emitted by the task class and catch the object which is also emitted (in my example, I emit only the result layer object).

Any layer which is created in a background thread should be transferred to the main thread before it is accessed, however I have noticed unexpected behavior even when a layer is passed to the main thread via a pyqtsignal and loaded directly into the project. For example, feature selection doesn't quite work as expected (selection/ deselection only happens when the canvas is refreshed e.g. by panning or zooming). Theoretically, the way to deal with this should be to move the layer object to the main thread before emitting the signal like so:

self.result_layer.moveToThread(QCoreApplication.instance().thread())

However, I didn't find that this made any difference in the example below. One workaround which I found to be successful is simply adding a call to the clone() method when passing the result layer to the addMapLayer() method.

Full example below:

from qgis.PyQt.QtCore import QSettings, QTranslator, QCoreApplication
from qgis.PyQt.QtGui import *
from qgis.PyQt.QtWidgets import *

from datetime import datetime
import os.path, os, shutil
from qgis.core import *
from qgis.gui import *
from qgis.utils import *
import processing
from PyQt5.QtCore import pyqtSignal


class MyTask(QgsTask):
    messaging = pyqtSignal(str)
    result = pyqtSignal(QgsVectorLayer)

    def __init__(self, desc, alg, params, task_no):
        self.desc = desc
        self.alg = alg
        self.params = params
        self.task_no = task_no
        self.result_layer = None
        QgsTask.__init__(self, self.desc)
        

    def run(self):
        start = datetime.now()
        self.result_layer = processing.run(self.alg, self.params)['OUTPUT']
        end = datetime.now()
        self.messaging.emit(self.desc)
        self.messaging.emit(
            f"Time required to achieve task {self.task_no}: {str((end - start).total_seconds())} seconds\n")
        return True
        
    def finished(self, result):
        if result:
#            self.result_layer.moveToThread(QCoreApplication.instance().thread())
            self.result.emit(self.result_layer)
                


class plugintestDialog(QDialog):
    def __init__(self, parent=None):
        super(plugintestDialog, self).__init__(parent)
        self.lyt = QVBoxLayout(self)
        self.cb = QgsMapLayerComboBox(self)
        self.pushButton = QPushButton("Test 1", self)
        self.progressbar = QProgressBar(self)
        self.progressbar.setMinimum(0)
        self.lyt.addWidget(self.cb)
        self.lyt.addWidget(self.pushButton)
        self.lyt.addWidget(self.progressbar)


class plugintest:

    messaging = pyqtSignal(str)

    def __init__(self, iface):
        self.iface = iface
        self.dlg = plugintestDialog()
        self.dlg.setMinimumWidth(350)
        self.dlg.cb.setFilters(QgsMapLayerProxyModel.LineLayer | QgsMapLayerProxyModel.PolygonLayer)
        self.dlg.pushButton.clicked.connect(self.centroid)

    def run(self):
        self.dlg.show()
        result = self.dlg.exec_()

    def centroid(self):
        self.dlg.pushButton.setEnabled(False)
        self.dlg.progressbar.setMaximum(0)
        layer = self.dlg.cb.currentLayer()

        centroid_params = {
        'ALL_PARTS': True,
        'INPUT': layer,
        'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
        }
        
        # create task 1
        self.centroid_task = MyTask('Creating Centroids...', 'native:centroids', centroid_params, 1)
        # connect signals for task 1
        self.centroid_task.messaging.connect(lambda t: print(t))
        self.centroid_task.result.connect(lambda d: self.buffering(d))
        # add task 1 to task manager
        QgsApplication.taskManager().addTask(self.centroid_task)

    def buffering(self, layer):
        buffer_params = {
        'DISSOLVE': True,
        'DISTANCE': 100,
        'END_CAP_STYLE': 0,
        'INPUT': layer,
        'JOIN_STYLE': 0,
        'MITER_LIMIT': 2,
        'SEGMENTS': 5,
        'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
        }
        # create task 2
        self.buffer_task = MyTask('Creating Buffers...', 'native:buffer', buffer_params, 2)
        # connect signals for task 2
        self.buffer_task.messaging.connect(lambda t: print(t))
        self.buffer_task.result.connect(self.load_layer)
        self.buffer_task.taskCompleted.connect(self.buffering_completed)
        # add task 2 to task manager
        QgsApplication.taskManager().addTask(self.buffer_task)

    def load_layer(self, res_lyr):
        #Using .clone() seems to avoid unexpected behaviour exhibited when
        #loading the result layer directly into the project
        QgsProject.instance().addMapLayer(res_lyr.clone())

    def buffering_completed(self):
        self.dlg.progressbar.setMaximum(1)
        self.buffer_task.disconnect()
        self.dlg.pushButton.setEnabled(True)


my_instance = plugintest(iface)
my_instance.run()

A final point I would like to make is that, since the logic your plugin seems to be chaining together processing algorithms, I would suggest that creating a parent processing algorithm would be a better approach. for example you could create a model, export it as a Python script, and simply add to the processing toolbox, or even create a full Processing Plugin.