Part C - Defining Custom Results

Now that you’ve seen how to implement a custom worker, let’s see how to customize the results of your workers. In this example, we will create a new result type and refactor the CustomWorker from Part B so that it uses this new result type.

By default, the results of an AsyncWorker are contained within an AsyncWorkerResults object. This object has the following attributes:

  • id: The unique id of the worker

  • warnings: Custom warnings encountered by the worker

  • errors: Custom errors encountered by the worker

  • results_dict: Result values of the workers task that were passed to complete

For many use cases, this basic result type is good enough. However, to make your application more robust and to take full advantage of type-hinting, you can define a custom result type that has more descriptive attributes.

To define a custom result type, simply create a new class that inherits from AsyncWorkerResults and within the class define your custom results as attributes:

1class CustomWorkerResults(AsyncWorkerResults):
2    custom_result_1 = None
3    custom_result_2 = None

To allow your worker to use your new result type, assign the results attribute of your worker to a new instance of your custom results within the worker’s __init__ method:

1class CustomAsyncWorker(AsyncWorker):
2    def __init__(self, delay_seconds: int, cycles=4):
3        super(CustomAsyncWorker, self).__init__()
4        self.delay_seconds = delay_seconds
5        self.cycles = cycles
6        # Redefine the self.results attribute
7        self.results = CustomWorkerResults()

Now that the result type has been assigned, your worker can store result data directly into results. This is what the run() method looks now like:

 1class CustomAsyncWorker(AsyncWorker):
 2    ...
 3    def run(self):
 4        self.emit_start()
 5        progress = 5
 6        self.update_progress(progress, 'Starting Task')
 7        for ii in range(self.cycles):
 8            time.sleep(self.delay_seconds)
 9            progress += 90 / self.cycles
10            self.update_progress(progress, f'Progress message #{ii + 1}')
11
12        # Store the results directly into the attributes of self.results
13        self.results.custom_result_1 = 'result 1'
14        self.results.custom_result_2 = 'result 2'
15        self.complete()

Storing result values into a custom results attribute is the best way to emit results from an AsyncWorker.

Note

For flexibility, PySink still allows you to pass results as kwargs to complete(), even if you’ve defined a custom result type. In this scenario, the complete() will still package the results into the results_dict. However, it will also attempt to map the kwargs to the attributes of the custom result type if and only if the keywords match the attributes.

Just like we can access the new result attributes directly within run(), we can also access them directly within the completion callback. We no longer need to extract the values from a dictionary, and type-hinting makes the data extraction less prone to spelling mistakes and KeyErrors:

1def completion_callback(results: CustomWorkerResults):
2    print(f'\nWorker Complete!')
3    print(f'\tErrors: {results.errors}')
4    print(f'\tWarnings: {results.warnings}')
5    print(f'\tResult Attribute 1: {results.custom_result_1}')
6    print(f'\tResult Attribute 2: {results.custom_result_2}')
7    sys.exit()  # Exit the App event loop

Below is the full implementation of the worker with a custom result type:

 1from PySide6.QtWidgets import QApplication
 2from PySink import AsyncManager, AsyncWorker, AsyncWorkerProgress, AsyncWorkerResults
 3import sys
 4import time
 5
 6
 7# Define a class representing your result type, storing result values as attributes
 8class CustomWorkerResults(AsyncWorkerResults):
 9    custom_result_1 = None
10    custom_result_2 = None
11
12
13class CustomAsyncWorker(AsyncWorker):
14    def __init__(self, delay_seconds: int, cycles=4):
15        super(CustomAsyncWorker, self).__init__()
16        self.delay_seconds = delay_seconds
17        self.cycles = cycles
18        # Redefine the self.results attribute
19        self.results = CustomWorkerResults()
20
21    def run(self):
22        self.emit_start()
23        progress = 5
24        self.update_progress(progress, 'Starting Task')
25        for ii in range(self.cycles):
26            time.sleep(self.delay_seconds)
27            progress += 90 / self.cycles
28            self.update_progress(progress, f'Progress message #{ii + 1}')
29
30        # Store the results directly into the attributes of self.results
31        self.results.custom_result_1 = 'result 1'
32        self.results.custom_result_2 = 'result 2'
33        self.complete()
34
35
36# Function to be called whenever a worker's task has started
37def worker_started_callback(worker_id: str):
38    print(f'Worker with id {worker_id} has started its task\n')
39
40
41# Function to be called whenever progress is updated
42def progress_callback(progress: AsyncWorkerProgress):
43    print(f'Progress Received, value: {progress.value}, message: {progress.message}')
44
45
46# Function to be called when the worker is finished. Note that the results are now of type CustomWorkerResults.
47def completion_callback(results: CustomWorkerResults):
48    print(f'\nWorker Complete!')
49    print(f'\tErrors: {results.errors}')
50    print(f'\tWarnings: {results.warnings}')
51    print(f'\tResult Attribute 1: {results.custom_result_1}')
52    print(f'\tResult Attribute 2: {results.custom_result_2}')
53    sys.exit()  # Exit the App event loop
54
55
56def run_main():
57    app = QApplication()
58    #   Create the Async Manager
59    manager = AsyncManager()
60    #   Create the Worker and pass in the necessary values
61    worker = CustomAsyncWorker(delay_seconds=1, cycles=3)
62    #   Connect the Worker's signals to their callbacks
63    worker.signals.started.connect(worker_started_callback)
64    worker.signals.progress.connect(progress_callback)
65    worker.signals.finished.connect(completion_callback)
66    #   Start the Worker and App event loop
67    manager.start_worker(worker)
68    app.exec()
69
70
71run_main()

Running this script produces the following console output:

 1Worker with id 0b160ae0-f3b1-4191-852a-fdd1e9c1c76b has started its task
 2
 3Progress Received, value: 5, message: Starting Task
 4Progress Received, value: 35.0, message: Progress message #1
 5Progress Received, value: 65.0, message: Progress message #2
 6Progress Received, value: 95.0, message: Progress message #3
 7
 8Worker Complete!
 9    Errors: []
10    Warnings: []
11    Result Attribute 1: result 1
12    Result Attribute 2: result 2

In most use cases, if your custom AsyncWorker will complete with result values, it is recommended that a custom result type is created and implemented. This makes your code more understandable and less prone to bugs. In the final part of the Basics, you will see how to customize your workers even further by defining custom signals.