What is a task ?
A task
is a fundamental object used in any Brick
. It is one of the main object of the workflow that you can find in the playground.
A task
is a special function that works with Constellab
object. It contains a run
method to execute any code. It takes resources
as inputs and resources
as outputs. It also takes a configuration
object.
For example, the object TableTransposer
is a task
that takes a Table
as input, transpose it rand return the transposed table
as output.
Tasks can be connected to each other to be executed, this is what we called a protocol
.
How to create a task ?
To create a new task in your brick, you need to create a new class that:
- Extend the
Task
class (import from gws_core
) - Is decorated with the
@task_decorator
(import from gws_core
). - Define the
input_specs
, output_specs
and config_specs
Here is an example of our RobotMove
task. This task takes a Robot
resource as input, update its positions based on the configuration
and return the Robot
.
from gws_core import (ConfigParams, FloatParam, InputSpec, OutputSpec,
StrParam, Task, TaskInputs, TaskOutputs, task_decorator)
from .robot_resource import Robot # local import
@task_decorator("RobotMove", human_name="Move robot",
short_description="This task emulates a short moving step of the robot", hide=True)
class RobotMove(Task):
input_specs = {'robot': InputSpec(Robot, human_name="Robot", short_description="The robot to move")}
output_specs = {'robot': OutputSpec(Robot)}
config_specs = {
'moving_step': FloatParam(default_value=0.1, short_description="The moving step of the robot"),
'direction': StrParam( default_value="north", allowed_values=["north", "south", "east", "west"], short_description="The moving direction")
}
def run(self, params: ConfigParams, inputs: TaskInputs) -> TaskOutputs:
print(f"Moving {params.get_value('moving_step')}", flush=True)
robot: Robot = inputs['robot']
robot.move(direction=params.get_value('direction'), moving_step=params.get_value('moving_step'))
return {'robot': robot}
Run method
The run method is the method executed when the system runs the task. You must right your code logic for this task in this method.
The run method has 2 parameters
-
params
of type ConfigParams
: this is a dict
object that contains the values of the configuration provided by the user. See Configuration for more information. - inputs of type
TaskInputs
: this is a dict
object that contains all the resources provided as input of the task. See Input & Outputs
for more information.
The run method must return a TaskOutputs
which is a dictionary of resources.
The task will be executed in a protocol, so the outputs of one task can be connected to the one or more inputs of the next task. With this you can pass resources that a task generated to another task.
Inputs and Outputs
When we create a task we need to define what are the task IO (inputs and outputs) and the type of each IO. A task takes only Resource
as IO so the type provided in the IO must be Resource
. When the task is run the inputs are provided in the run
method.
To define the specifications of the task IO we must define the input_specs
and output_specs
.
Input
The input_specs
object is a dictionary of InputSpec
. The keys of the dictionary are important because it helps you retrieve the input in the run method. In the above example we retrieve the Robot
resource by calling inputs['robot']
robot
Is the key define in the input_specs
.
The InputSpec
object help you to define your inputs. It takes multiple parameters:
-
resource_types
: a resource type or a List of resource types. If 1 type is provided (recommended), the system only accepts this type and sub-classes for this input. If multiple types are provided, the system will accept any of this classes and sub-classes. Before running the task, the system checks the inputs and if 1 input is not compatible the task is not run. -
human_name
: pretty name that will be displayed in the playground. -
short_description
: small description to add information about this input (also for the playground). -
is_optional
: this input might not be connected to another task output and the task will still be executed. If the input is connected, the system will wait for the input to be provided before running the task. Also tells that None value is allowed as input.
Outputs
The output_specs
object is a dictionary of OutputSpec
. The keys of the dictionary are important because it must match the keys of the returned object in the run method. For example in the RobotMove
task, we returned : return {'robot': robot}
which means we return the resource robot
with the key robot
. The robot
key is the same as defined in the output_specs
of the task.
The OutputSpec
object help you to define your outputs. It takes multiple parameters (some like InputSpec
):
-
resource_types
: a resource type or a List of resource types. If 1 type is provided (recommended), it tells the system that this task returns a specific type(and sub-classes) for this output. If multiple it tells that the task will return one of the provided type (and sub-classes) for this output. -
human_name
: pretty name that will be displayed in the playground. -
short_description
: small description to add information about this output (also for the playground). -
is_optional
: tell that this output may return None or not being provided. -
is_constant
(for expert): When true, it tells the system that the output resource was not modified from the input resource. The system will not need to create a new resource after the task. -
sub_class
(for expert): When true, it tells that the resource_types are compatible with any child class of the provided resource type (useful for task that work with generic resource type).
Configuration
The config_specs
attribute allows you to expose configuration to the playground. You can define multiple parameters with different type and this will generate a form available in the playground when the user uses the task.
The config_specs
object is a dictionary of Param
. The keys are important because it helps you retrieve the parameters value from the params object (first parameter of the run method). In the above example we retrieve the two parameters’ values direction
and moving_step
by calling params.get_value('direction')
moving_step=params.get_value('moving_step')
. The params object is a python dict with additional methods.
There are different parameter types (int, float, bool, str..) but they all have common options when we define a new parameter:
-
human_name
: pretty name that will be displayed in the playground when configuring the task.
-
short_description
: small description to add information about this parameter (also for the playground).
-
default_value
: default value to use when the user did not provide a value for this parameter. If a default value is provided, it automatically set the optional to True. If not provided and the optional is set to False, the parameter is mandatory. If you want to set a default value to None, set optional to True.
-
optional
: define the parameters as optional. If default_value is not provided and the user did not define this parameter, the value will be None -
visibility
-
public
: main param visible in the first config section in the playground -
protected
: considered as advanced param, it will be in the advanced section in the playground. It must have a default value or be optional. -
private
(for expert) : it will not be visible in the playground. It must have a default value or be optional)
-
allowed_values
: list of allowed value for the parameter. It will be rendered as a select in the playground unit: define the measure unit nit of this parameter
Basic Params
Here is the list of all the basic param types with additional options for some of them:
-
StrParam
: string param. It will show a basic text input in the playground. Use allowed value to restrict possible values min_length: minimal length for the provided text max_length: maximal length for the provided text -
IntParam
& FloatParam
: numeric param. It will show an input with only number possible. -
min_value
: minimum value for the param -
max_value
: maximum value for the param
-
BoolParam
: Boolean param. It will show a checkbox. -
ListParam
: List param. Show a text input in the playground where the user can provide multiple values. The value for this param will be a list of strings. It also works with allowed_value, to allow user to select multiple values from a list of choices. -
DictParam
(for expert) : only to use for hidden parameters (it has no representation in the playground). -
TagsParams
: Show a special input to select multiple tags. The value of the parameter will be a list of tags.
ParamSet
Use to define a group of parameters that can be added multiple times. The ParamSet
contains a dictionary of parameters. Use this when there are 2 or more configurations that are related and you want the user to be able to provided multiple combinations of theses parameters.
When not optional
, the user will need to provide at least one value (a combination) in the playground.
Use the max_number_of_occurence
attribute to limit the number of combinations the user can provide.
The value of this parameter will be a list of dictionaires.
Task Special methods
Log messages
There are multiple method that exist to write logs during the task. The logs are available in the playground and are update in real time. The logs are really usefull when the task takes some time and it is also possible to log progress.
Here are the log methods:
-
update_progress_value
: log a progress message and update the progress value (value from 0 to 100). Ex: update_progress_value(10, 'Reached 10%')
-
log_info_message
: log an information message -
log_success_message
: log a success message -
log_warning_message
: log a warning message -
log_error_message
: log an error message -
log_message
: log a message with the type in parameter
You can call theses methods in the run methods like this : self.log_info_message('Hello world')
.
Check before run
The check_before_run
method can be overwritten to perform custom check before running task. It must return a CheckBeforeTaskResult
that contains 2 objects :
-
result
: boolean - If
True
, everything is ok - If
False
, the task will not be executed after this check it might be run later if they are some SKippableIn inputs. If all the input values were provided and the check retuns False. the task will endup in error because it won't be run
-
message
: Optional[str] - If
False
, a message can be provided to log the error message if the task will not be called
Run after task
The run_after_task
method can be overwritten to perform action after the task run. This method is called after the resource save. This method is useful to delete temporary objects (like files) to clear the server after the task is run.
Create temp directory
The create_tmp_dir
method can be called inside the task to create a temporary directory that will be automatically deleted after the task is run (on the run_after_task
method). Output files and folders of the task are moved out of this directory before it is deleted.
Test the task
There are 2 ways of testing the task and we recommend to do both of them:
- Write unit test
- Test in dev environment
Write unit test
Writing a unit test in the best and easiest way to test your task and make sure it still works after some changes (you change it or update a dependency for example).
Create a new test file in the tests folder. The name of the file must start with test_
.
Here is a example on how to test the RobotMove
task.
from gws_core import TaskRunner
from gws_core.impl.robot.robot_resource import \
Robot # local import of the resource to test
from gws_core.impl.robot.robot_tasks import \
RobotMove # local import of the task to test
from unittest import TestCase

class TestRobotMove(TestCase):
def test_robot_move(self):
# create an empty robot with position [0,0]
robot_output = Robot.empty()
# create a task runner and configure it, then run it
runner = TaskRunner(
task_type=RobotMove,
params={'moving_step': 3, 'direction': 'south'},
inputs={'robot': robot_output},
)
outputs = await runner.run()
# retrieve the robot output
robot_output: Robot = outputs['robot']
# check the robot position, y should be -3 as he went south
self.assertEqual(robot_output.position, [0, -3])
Let's break down this example.
Firstly we declared the class TestRobotMove
that extend the TestCase
. The test class must extend the TestCase
to be considered as a test file by VsCode.
We manually create our Robot
resource.
The TaskRunner
is the main part of this test. This class allows you to run a task manually with custom config and input and retrieve the outputs. Then we create the task runner by passing the task type RobotMove
, the configuration we want to test, and the input (the previously created robot).
Then we run the task by calling the runner.run()
, this method will return a TaskOutputs
. We can then retrieve the robot from the outputs and check that the value of the robot is correct. Here we check that the position of the robot was changed.
To learn how to run the test and debug it, see the following link.
Dev environment > debug-tests
You can also test the check_before_run
method with the task runner. It may also be necessary to call the run_after_task
at the end of your test if your task generated side data that need to be cleared.
Test in dev environment
Once you unit tests are ok, you can test your task in the dev environment. See the following link:
Dev environment > start-dev-server
Document the task
Once your task is developed and tested, we recommend to document it. Here is a list of things you can do to improve the documentation of your task:
- Define the
human_name
and short_description
in the @task_decorator
- Define the
human_name
and short_description
in the input_specs
, output_specs
and config_specs
- Write comment in the python code of your classe.
You can write complete documentation using python comment. It supports the markdown syntax: (https://www.markdownguide.org/basic-syntax/). This documentation is really important and it will be available in Constellab community and in the playground. Here is an example on how to do it.
class RobotMove(Task):
"""
This is a documentation that support **markdown** syntax.
And this documentation will be available in
- the hub
- the playground.
"""
Deploy your task
Once you created your test and write your documentation, you can deploy your task in a new version of your brick. To do so, check the following link: Deploy a brick version
Use external data
If your task rely on one or more external data (like a database) that needs to be downloaded first you can use the TaskFileDownloader
. This class let you download external file resource (using http protocol) and store it in a specific folder for your brick.
To create the TaskFileDownloder from your task simple do the following :
# create the file_downloader from a task.
file_downloader = TaskFileDownloader(MyTaskClass.get_brick_name(), self.message_dispatcher)
# download a file
file_path = file_downloader.download_file_if_missing("my_file_url", "file_name.json")
The external downloaded data can be viewed directly in the lab in the Monitoring
> Other
> Brick data
.
In this page you can delete manually a data so it will be force to be re-downloaded next time the task is executed.