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)
start_processing the task - encapsulated in a generic worker object - gets moved to a thread and the processing gets started.
my_thread = QThread() my_thread.start() def start_processing(): my_worker = GenericWorker(task) my_worker.moveToThread(my_thread) 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" return_value = per_file_task(some_arguments, some_kw_arguments) # do some post processing with return value return task_result
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
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" return_value = per_file_task(some_arguments, some_kw_arguments) # do some post processing with return value if progress_callback: proggress_callback(i, len(filelist)) return task_result
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?