robocorp-workitems

Reserving and releasing input items

When an execution in Control Room starts, the first input item is automatically reserved. This first item is also loaded by the library when the task execution starts.

After an item has been handled, it should be released as either passed or failed. There can only be one reserved input item at a time.

Reserving can be done explicitly by calling the reserve method, which also acts as a context manager:

with workitems.inputs.reserve() as item: print("Handling item!")

Another option is to loop through all inputs, which implicitly reserves and releases the corresponding items:

for item in workitems.inputs: print("Handling item!")

Releasing can also be done explicitly to set specific errors, or to mark items as done before exiting the block:

for item in workitems.inputs: order_id = item.payload["order_id"] if not is_valid_id(order_id): item.fail(code="INVALID_ID", message=f"Invalid order id: {order_id}") ... item.done()

Creating outputs

For each input work item, you can create any amount of output work items. These will be forwarded to the next step of a process, or set as the final result if there are no further steps.

In most cases, it's enough to create an output item directly using the outputs object:

workitems.outputs.create( payload={"key": "value"}, files=["path/to/file.txt"], )

Internally, Control Room keeps a relationship between the parent and child work items. The above example always uses the currently reserved item as the parent, but it's also possible to create an output explicitly from the input item:

with workitems.inputs.reserve() as item: out = item.create_output() out.payload = {"key": "value"} out.save()

In some cases, it's also useful to create an output item and then modify it before saving:

item = workitems.outputs.create(save=False) for index, path in enumerate(directory.glob("*.pdf")): item.add_file(path, name=f"document-{index}") item.save()

Local development

Using the VSCode extension

If you are developing in VSCode with the Robocorp Code extension, you can utilize the built in local development features described in the Developing with work items locally section of the Using work items development guide.

This allows you to develop and test your work items before deploying to Control Room.

Using a custom editor

It's also possible to develop locally with a custom editor, but it requires some configuration.

To enable the development mode for the library, you should set the environment variable RC_WORKITEM_ADAPTER with the value FileAdapter. This tells the library to use local files for simulating input and output queues for work items, in the form of JSON files.

The environment variables RC_WORKITEM_INPUT_PATH and RC_WORKITEM_OUTPUT_PATH should also be set, and should contain the paths to the input and output JSON files. The output file will be created by the library, but the input file should be created manually.

An example of an input file with one work item:

[ { "payload": { "variable1": "a-string-value", "variable2": ["a", "list", "value"] }, "files": { "file1": "path/to/file.ext" } } ]

The format of the file is a list of objects, where each item in the list corresponds to one work item. The payload key can contain any arbitrary JSON, and the files key is pairs of names and paths to files.

Email triggering

A process can be started in Control Room by sending an email, after which the payload and files will contain the email metadata, text, and possible attached files. This requires the Parse email configuration option to be enabled for the process in Control Room.

An input work item in this library has a helper method called email(), which can be used to parse an email into a typed container:

from robocorp import workitems from robocorp.tasks import task @task def read_email(): item = workitems.inputs.current email = item.email() print("Email sent by:", email.from_.address) print("Email subject:", email.subject) payload = json.loads(email.text) print("Received JSON payload:", payload)

Email structure

The email content in the payload is parsed into the following typed container:

class Address: name: str address: str class Email: from_: Address to: list[Address] cc: list[Address] bcc: list[Address] subject: str date: datetime reply_to: Optional[Address] message_id: Optional[str] text: Optional[str] html: Optional[str] errors: list[str]

Error handling

When Control Room can not parse the email, for example when the content is too large to handle, it will set specific error fields. These are parsed by the work items library and raised as a corresponding exception. These errors can also be ignored with the ignore_errors argument to the email() method.

Using the raw payload

In some use-cases, it's necessary to access additional metadata such as custom headers. Control Room attaches the raw unparsed email as a file attachment, which can be parsed by the Robot code. One option for this is the Python built-in email library:

import email from robocorp import workitems from robocorp.tasks import task @task def parse_message_id(): for item in workitems.inputs: path = item.get_file("__raw_mail") with open(path) as fd: message = email.message_from_file(fd) message_id = message["Message-ID"]

Further documentation

To learn more about email triggering in Control Room, see the docs page.

Collecting all inputs

Sometimes it's necessary to reduce multiple input work items into one output work item, e.g. for reporting purposes. The tricky part is that there needs to be a reserved input work item to create an output, and there is an unknown amount of input work items. This means that it's not possible to first loop all inputs and then create the output.

One way to solve this is to create an output from the first item, and then modify it later:

from robocorp import workitems from robocorp.tasks import task @task def summarize(): output = workitems.outputs.create() errors = [] for item in workitems.inputs: if error := item.payload.get("error"): errors.append(error) output.payload = {"errors": errors} output.save()

Flow control with exceptions

Failures for input work items can be set explicitly with the fail() method, which takes in an exception type (e.g. BUSINESS or APPLICATION), and an optional custom error code and human-readable message.

However, sometimes complicated business logic needs to fail the entire input item within a function that does not have direct access to it. For this use-case the library exposes two exceptions: BusinessException and ApplicationException.

When the library catches either of these exceptions, it knows to use it to set the failure state of the work item accordingly.

Example usage

from robocorp import workitems from robcorp.tasks import task @task def handle_transactions(): with workitems.inputs.current as item: handle_transaction(item.payload) def handle_transaction(data): check_valid_id(data["transactionId"]) ... def check_valid_id(value): if len(value) < 8: raise workitems.BusinessException( code="INVALID_ID", message="Transaction ID length is too short!", )

Error boundaries

Where the exceptions are caught depends on the calling code, as there are several boundaries where it can happen.

Task end

The outermost boundary is the task itself. If a library exception is thrown from within the task and no other code catches it, the values are used to set the input work item's state before the task fails.

from robocorp import workitems from robcorp.tasks import task @task def task_boundary(): raise workitems.ApplicationException("Oops, always fails")

Note: The task itself will always fail when an exception is caught on this level.

Context manager

A reserved input item can be used as a context manager, which automatically sets the pass or fail state depending on whether an exception happens before exiting the block or not.

from robocorp import workitems from robcorp.tasks import task @task def context_boundary(): # Sets the work item's state as completed after exiting the block with workitems.inputs.current: pass # Sets the work item's state as failed when catching the exception with workitems.inputs.reserve(): raise workitems.ApplicationException("Something went wrong") print("This does not run") print("This runs")

Note: When the context manager catches a work item exception, it continues without failing the entire task.

Looping inputs

While iterating the input queue automatically reserves and releases items, it can't catch the exceptions while iterating. If these exceptions are used in a looping task, they should be combined as follows:

import random from robocorp import workitems from robcorp.tasks import task @task def loop_boundary(): # Loops through all work items, even though some of them throw an exception for item in workitems.inputs: with item: if random.choice([True, False]): raise workitems.ApplicationException("Random failure")

Modifying inputs

Although not recommended, it's possible to modify input work items. As an example, a process might need to store additional metadata for failed items.

This is potentially dangerous, as it will reduce traceability of the process and maybe even make it behave in unexpected ways. If a modified item is retried, it could result in something that was not intended.

With these caveats, modifying an input item is relatively easy:

from robocorp import workitems from robocorp.tasks import task @task def clear_payload(): # NOTE: This is not recommended item = workitems.inputs.current item.payload = None item.save()

This example would clear all payloads from input work items.