I have some PySide GUI application which is capable of executing long running tasks.

The targets of my application design is to

• put the task execution to a thread to not block the GUI (e.g. for pausing the task)
• display the task progress in a progress bar in the GUI
• be able to call a task method separately from my py.test envorionment

My current setup has a start_processing method, which starts a task depending on the inputs of the GUI. I am using a generic worker object in combination with QThread as each task has a different set of arguments and i dont want to write a specialised thread subclass for each task.

class GenericWorker(QObject):
start = QtCore.Signal(str)
progress_changed = QtCore.Signal((int, int))
finished = QtCore.Signal(object)

def __init__(self, function, *args, **kwargs):
super(GenericWorker, self).__init__()

self.function = function
self.args = args
self.kwargs = kwargs
self.start.connect(self.run)

def run(self, some_string_arg):
try:
return_value = self.function(*self.args, **self.kwargs)
except Exception as e:
# log exception
else:
if return_value is not None:
finished.emit(return_value)
else:
finished.emit(0)


In start_processing the task - encapsulated in a generic worker object - gets moved to a thread and the processing gets started.

my_thread = QThread()

def start_processing():
my_worker.progress_changed.connect(update_progressbar)
my_worker.start.emit("hello")

def update_progressbar(i, total):
# set progress bar to the corresponding percentage


In almost all use cases a task looks like that:

def task(some_arguments, some_kw_arguments):
for filepath in filelist:
# do some preparation for the actual "per file task"
# do some post processing with return value



To be able to report the progress of a task, i want to emit a signal from the generic worker (running in its thread) after each per_file_task which states the current progress value. As the task is not defined inside the generic worker class but outside in some Processing class, i need to connect the signal of the generic worker to a update_progressbar method of the GUI.

My idea is to define a callback function in each task which gets called after each per_file_task.

def task(some_arguments, some_kw_arguments, progress_callback=None):
for i, filepath in enumerate(filelist):
# do some preparation for the actual "per file task"
# do some post processing with return value
if progress_callback:
proggress_callback(i, len(filelist))



Now that proggress_callback function has to get populated after the task has been registered with the generic worker thread.

def __init__(self, function, *args, **kwargs):
super(GenericWorker, self).__init__()

self.function = function
self.args = args
self.kwargs = kwargs
self.kwargs["progress_callback"] = self.progress_changed.emit
self.start.connect(self.run)


But for me this doesn't look really clean.

My first attempt was to then to split up the processing of the task's for loop into registering only the per file task with the GenericWorker and update its filepath argument for every file. So the update of the progress bar could be done outside of the thread. But that would break my requirement for the testability as it introduces side effects into the task's logic.

So my final question is: How could i realize this in a cleaner way? Or are there other patterns which are commonly used in such situations?

• Just had the idea I could use a generator for the task function and yield the current progress after Ever per file task. Would be cleaner from my point of view – maggie Oct 8 '17 at 5:04
• But i still need to check how to return the value of the task function then. Perhaps via the StopIteration Exception? – maggie Oct 8 '17 at 14:08