Python API

class RPA.Robocorp.WorkItems.WorkItems(autoload: bool = True, root: ~typing.Optional[str] = None, default_adapter: ~typing.Union[~typing.Type[~RPA.Robocorp.WorkItems.BaseAdapter], str] = <class 'RPA.Robocorp.WorkItems.RobocorpAdapter'>, auto_parse_email: ~typing.Optional[~typing.Dict[~typing.Union[str, ~typing.Tuple[str, ...]], ~typing.Union[str, ~typing.Tuple[str, ...]]]] = {('email.text', '__mail.html'): ('email.body', 'parsedEmail.Body'), 'rawEmail': 'parsedEmail'})

A library for interacting with Control Room work items.

Work items are used for managing data that go through multiple steps and tasks inside a process. Each step of a process receives input work items from the previous step, and creates output work items for the next step.

Item structure

A work item’s data payload is JSON and allows storing anything that is serializable. This library by default interacts with payloads that are a dictionary of key-value pairs, which it treats as individual variables. These variables can be exposed to the Robot Framework task to be used directly.

In addition to the data section, a work item can also contain files, which are stored by default in Robocorp Control Room. Adding and using files with work items requires no additional setup from the user.

Loading inputs

The library automatically loads the first input work item, if the library input argument autoload is truthy (default).

After an input has been loaded its payload and files can be accessed through corresponding keywords, and optionally these values can be modified.

E-mail triggering

Since a process can be started in Control Room by sending an e-mail, a body in Text/JSON/YAML/HTML format can be sent as well and this gets attached to the input work item with the rawEmail payload variable. This library automatically parses the content of it and saves into parsedEmail the dictionary transformation of the original e-mail.

If “Parse email” Control Room configuration option is enabled (recommended), then your e-mail is automatically parsed in the work item under the email payload variable, which is a dictionary containing a body holding the final parsed form of the interpreted e-mail body. The payload variable parsedEmail is still available for backwards compatibility reasons and holds the very same body inside the parsedEmail[Body].

E-mail attachments will be added into the work item as files. Read more on: https://robocorp.com/docs/control-room/attended-or-unattended/email-trigger

Example:

After starting the process by sending an e-mail with a body like:

{
    "message": "Hello world!"
}

The robot can use the parsed e-mail body’s dictionary:

*** Tasks ***
Using Parsed Emails
    ${mail} =    Get Work Item Variable    email
    Set Work Item Variables    &{mail}[body]
    ${message} =     Get Work Item Variable     message
    Log    ${message}    # will print "Hello world!"

The behaviour can be disabled by loading the library with auto_parse_email=${None} or altered by providing to it a dictionary with one “key: value” where the key is usually “email.text” (deprecated “rawEmail”, the variable set by Control Room, which acts as source for the parsed (deprecated raw) e-mail data) and the value can be “email.body” (deprecated “parsedEmail”, where the parsed e-mail data gets stored into), value which can be customized and retrieved with Get Work Item Variable.

Creating outputs

It’s possible to create multiple new work items as an output from a task. With the keyword Create Output Work Item a new empty item is created as a child for the currently loaded input.

All created output items are sent into the input queue of the next step in the process.

Active work item

Keywords that read or write from a work item always operate on the currently active work item. Usually that is the input item that has been automatically loaded when the execution started, but the currently active item is changed whenever the keywords Create Output Work Item or Get Input Work Item are called. It’s also possible to change the active item manually with the keyword Set current work item.

Saving changes

While a work item is loaded automatically when a suite starts, changes are not automatically reflected back to the source. The work item will be modified locally and then saved when the keyword Save Work Item is called. This also applies to created output work items.

It is recommended to defer saves until all changes have been made to prevent leaving work items in a half-modified state in case of failures.

Local Development

While Control Room is the default implementation, it can also be replaced with a custom adapter. The selection is based on either the default_adapter argument for the library, or the RPA_WORKITEMS_ADAPTER environment variable. The library has a built-in alternative adapter called FileAdapter for storing work items to disk.

The FileAdapter uses a local JSON file for input work items. It’s a list of work items, each of which has a data payload and files.

An example of a local file with one work item:

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

Output work items (if any) are saved to an adjacent file with the same name, but with the extension .output.json. You can specify through the “RPA_OUTPUT_WORKITEM_PATH” env var a different path and name for this file.

Simulating the Cloud with Robocorp Code 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.

Examples

Robot Framework

In the following example a task creates an output work item, and attaches some variables to it.

*** Settings ***
Library    RPA.Robocorp.WorkItems

*** Tasks ***
Save variables to Control Room
    Create Output Work Item
    Set work item variables    user=Dude    mail=address@company.com
    Save Work Item

In the next step of the process inside a different robot, we can use previously saved work item variables. Also note how the input work item is loaded implicitly when the suite starts.

*** Settings ***
Library    RPA.Robocorp.WorkItems

*** Tasks ***
Use variables from Control Room
    Set task variables from work item
    Log    Variables are now available: s${user}, ${mail}

Python

The library can also be used through Python, but it does not implicitly load the first work item.

import logging
from RPA.Robocorp.WorkItems import WorkItems

def list_variables(item_id):
    library = WorkItems()
    library.get_input_work_item()

    variables = library.get_work_item_variables()
    for variable, value in variables.items():
        logging.info("%s = %s", variable, value)

EMAIL_BODY_LOADERS = [('JSON', <function loads>), ('YAML', <function full_load>)]

ROBOT_AUTO_KEYWORDS = False

ROBOT_LIBRARY_DOC_FORMAT = 'REST'

ROBOT_LIBRARY_SCOPE = 'GLOBAL'

ROBOT_LISTENER_API_VERSION = 2

property active_input: Optional[WorkItem]

property adapter

add_work_item_file(path, name=None)

Add given file to work item.

Parameters

• path – Path to file on disk

• name – Destination name for file. If not given, current name of local file is used.

NOTE: Files are not uploaded before work item is saved

Example:

*** Tasks ***
Example task
    Add work item file    output.xls
    Save Work Item

add_work_item_files(pattern)

Add all files that match given pattern to work item.

Parameters

pattern – Path wildcard pattern

Example:

*** Tasks ***
Example task
    Add work item files    %{ROBOT_ROOT}/generated/*.csv
    Save Work Item

clear_work_item()

Remove all data and files in the current work item.

Example:

*** Tasks ***
Clearing a work item
    Clear work item
    Save work item

from RPA.Robocorp.WorkItems import WorkItems

wi = WorkItems()
wi.get_input_work_item()
wi.clear_work_item()
wi.save_work_item()

create_output_work_item(variables: Optional[dict] = None, files: Optional[Union[str, List[str]]] = None, save: bool = False) → WorkItem

Create a new output work item with optional variables and files.

An output work item is always created as a child for an input item, therefore a non-released input is required to be loaded first. All changes to the work item are done locally and are sent to the output queue after the keyword Save Work Item is called only, except when save is True.

Parameters

• variables – Optional dictionary with variables to be set into the new output work item.

• files – Optional list or comma separated paths to files to be included into the new output work item.

• save – Automatically call Save Work Item over the newly created output work item.

Returns

The newly created output work item object.

Examples

Robot Framework

*** Tasks ***
Create output items with variables then save
    ${customers} =  Load customer data
    FOR     ${customer}    IN    @{customers}
        Create Output Work Item
        Set Work Item Variables    id=${customer.id}
        ...     name=${customer.name}
        Save Work Item
    END

Create and save output items with variables and files in one go
    ${customers} =  Load customer data
    FOR     ${customer}    IN    @{customers}
        &{customer_vars} =    Create Dictionary    id=${customer.id}
        ...     name=${customer.name}
        Create Output Work Item     variables=${customer_vars}
        ...     files=devdata${/}report.csv   save=${True}
    END

Python

from RPA.Robocorp.WorkItems import WorkItems

wi = WorkItems()
wi.get_input_work_item()
customers = wi.get_work_item_variable("customers")
for customer in customers:
    wi.create_output_work_item(customer, save=True)

property current: WorkItem

delete_work_item_variables(*names, force=True)

Delete variable(s) from the current work item.

Parameters

• names – Names of variables to remove

• force – Ignore variables that don’t exist in work item

Example:

*** Tasks ***
Example task
    Delete work item variables    username    email
    Save Work Item

for_each_input_work_item(keyword_or_func: Union[str, Callable], *args, items_limit: int = 0, return_results: bool = True, **kwargs) → List[Any]

Run a keyword or function for each work item in the input queue.

Automatically collects and returns a list of results, switch return_results to False for avoiding this.

Parameters

• keyword_or_func – The RF keyword or Py function you want to map through all the work items

• args – Variable list of arguments that go into the called keyword/function

• kwargs – Variable list of keyword arguments that go into the called keyword/function

• items_limit – Limit the queue item retrieval to a certain amount, otherwise all the items are retrieved from the queue until depletion

• return_results – Collect and return a list of results given each keyword/function call if truthy

Example:

*** Tasks ***
Log Payloads
    @{lengths} =     For Each Input Work Item    Log Payload
    Log   Payload lengths: @{lengths}

*** Keywords ***
Log Payload
    ${payload} =     Get Work Item Payload
    Log To Console    ${payload}
    ${len} =     Get Length    ${payload}
    [Return]    ${len}

OR

import logging
from RPA.Robocorp.WorkItems import WorkItems

library = WorkItems()

def log_payload():
    payload = library.get_work_item_payload()
    print(payload)
    return len(payload)

def log_payloads():
    library.get_input_work_item()
    lengths = library.for_each_input_work_item(log_payload)
    logging.info("Payload lengths: %s", lengths)

log_payloads()

get_current_work_item() → WorkItem

Get the currently active work item.

The current work item is used as the target by other keywords in this library.

Keywords Get Input Work Item and Create Output Work Item set the active work item automatically, and return the created instance.

With this keyword the active work item can be retrieved manually.

Example:

*** Tasks ***
Example task
    ${input} =    Get Current Work Item
    ${output} =   Create Output Work Item
    Set Current Work Item    ${input}

get_input_work_item(_internal_call: bool = False) → WorkItem

Load the next work item from the input queue, and set it as the active work item.

Each time this is called, the previous input work item is released (as DONE) prior to reserving the next one. If the library import argument autoload is truthy (default), this is called automatically when the Robot Framework suite starts.

get_work_item_file(name, path=None) → str

Get attached file from work item to disk. Returns the absolute path to the created file.

Parameters

• name – Name of attached file

• path – Destination path of file. If not given, current working directory is used.

Example:

*** Tasks ***
Example task
    ${path}=    Get work item file    input.xls
    Open workbook    ${path}

get_work_item_files(pattern, dirname=None) → List[str]

Get files attached to work item that match given pattern. Returns a list of absolute paths to the downloaded files.

Parameters

• pattern – Filename wildcard pattern

• dirname – Destination directory, if not given robot root is used

Example:

*** Tasks ***
Example task
    ${paths}=    Get work item files    customer_*.xlsx
    FOR  ${path}  IN  @{paths}
        Handle customer file    ${path}
    END

get_work_item_payload()

Get the full JSON payload for a work item.

NOTE: Most use cases should prefer higher-level keywords.

Example:

*** Tasks ***
Example task
    ${payload}=    Get work item payload
    Log    Entire payload as dictionary: ${payload}

get_work_item_variable(name, default=<object object>)

Return a single variable value from the work item, or default value if defined and key does not exist.

If key does not exist and default is not defined, raises KeyError.

Parameters

• name – Name of variable

• default – Default value if key does not exist

Robot Framework Example:

*** Tasks ***
Using a work item
    ${username}=    Get work item variable    username    default=guest

Python Example:

from RPA.Robocorp.WorkItems import WorkItems

wi = WorkItems()
wi.get_input_work_item()
customers = wi.get_work_item_variable("customers")
print(customers)

get_work_item_variables()

Read all variables from the current work item and return their names and values as a dictionary.

Robot Framework Example:

*** Tasks ***
Example task
    ${variables}=    Get work item variables
    Log    Username: ${variables}[username], Email: ${variables}[email]

Python Example:

from RPA.Robocorp.WorkItems import WorkItems wi = WorkItems() wi.get_input_work_item() input_wi = wi.get_work_item_variables() print(input_wi[“username”]) print(input_wi[“email”])

inputs: List[WorkItem]

Input work items

list_work_item_files()

List the names of files attached to the current work item.

Example:

*** Tasks ***
Example task
    ${names}=    List work item files
    Log    Work item has files with names: ${names}

list_work_item_variables()

List the variable names for the current work item.

Example:

*** Tasks ***
Example task
    ${variables}=    List work item variables
    Log    Available variables in work item: ${variables}

outputs: List[WorkItem]

Output work items

release_input_work_item(state: Union[State, str], exception_type: Optional[Union[Error, str]] = None, code: Optional[str] = None, message: Optional[str] = None, _internal_release: bool = False)

Release the lastly retrieved input work item and set its state.

This can be released with DONE or FAILED states. With the FAILED state, an additional exception can be sent to Control Room describing the problem that you encountered by specifying a type and optionally a code and/or message. After this has been called, no more output work items can be created unless a new input work item has been loaded again.

Parameters

• state – The status on the last processed input work item

• exception_type – Error type (BUSINESS, APPLICATION). If this is not specified, then the cloud will assume UNSPECIFIED

• code – Optional error code identifying the exception for future filtering, grouping and custom retrying behaviour in the cloud

• message – Optional human-friendly error message supplying additional details regarding the sent exception

Example:

*** Tasks ***
Example task
    Login into portal
        ${user} =     Get Work Item Variable    user
        ${doc} =    Get Work Item Variable    doc
        TRY
            Login Keyword    ${user}
            Upload Doc Keyword    ${doc}

        EXCEPT    Login Failed
            Release Input Work Item     FAILED
            ...    exception_type=APPLICATION
            ...    code=LOGIN_PORTAL_DOWN
            ...    message=Unable to login, retry again later.

        EXCEPT    Format Error    AS    ${err}
            ${message} =    Catenate
            ...    Document format is not correct and cannot be uploaded.
            ...    Correct the format in this work item and try again.
            ...    Full error message received: ${err}
            Release Input Work Item     FAILED
            ...    exception_type=BUSINESS
            ...    code=DOC_FORMAT_ERROR
            ...    message=${message}

        END

OR

from RPA.Robocorp.WorkItems import State, WorkItems

library = WorkItems()

def process_and_set_state():
    library.get_input_work_item()
    library.release_input_work_item(State.DONE)
    print(library.current.state)  # would print "State.DONE"

process_and_set_state()

remove_work_item_file(name, missing_ok=True)

Remove attached file from work item.

Parameters

• name – Name of attached file

• missing_ok – Do not raise exception if file doesn’t exist

NOTE: Files are not deleted before work item is saved

Example:

*** Tasks ***
Example task
    Remove work item file    input.xls
    Save Work Item

remove_work_item_files(pattern, missing_ok=True)

Removes files attached to work item that match the given pattern.

Parameters

• pattern – Filename wildcard pattern

• missing_ok – Do not raise exception if file doesn’t exist

Example:

*** Tasks ***
Example task
    Remove work item files    *.xlsx
    Save Work Item

root

Variables root object in payload

save_work_item()

Save the current data and files in the work item. If not saved, all changes are discarded when the library goes out of scope.

set_current_work_item(item: WorkItem)

Set the currently active work item.

The current work item is used as the target by other keywords in this library.

Keywords Get Input Work Item and Create Output Work Item set the active work item automatically, and return the created instance.

With this keyword the active work item can be set manually.

Robot Framework Example:

*** Tasks ***
Creating outputs
    ${input}=    Get Input Work Item
    ${output}=   Create Output Work Item
    Set current work item    ${input}

Python Example:

from RPA.Robocorp.WorkItems import WorkItems

wi = WorkItems()
parent_wi = wi.get_input_work_item()
child_wi = wi.create_output_work_item()
wi.set_current_work_item(parent_wi)

set_task_variables_from_work_item()

Convert all variables in the current work item to Robot Framework task variables, see variable scopes.

Example:

*** Tasks ***
Example task
    # Work item has variable INPUT_URL
    Set task variables from work item
    Log    The variable is now available: ${INPUT_URL}

set_work_item_payload(payload)

Set the full JSON payload for a work item.

Parameters

payload – Content of payload, must be JSON-serializable

NOTE: Most use cases should prefer higher-level keywords. Using this keyword may cause errors when getting the payload via the normal Get work item variable and Get work item variables keywords if you do not set the payload to a dict.

Example:

*** Tasks ***
Example task
    ${output}=    Create dictionary    url=example.com    username=Mark
    Set work item payload    ${output}

set_work_item_variable(name, value)

Set a single variable value in the current work item.

Parameters

• name – Name of variable

• value – Value of variable

Robot Framework Example:

*** Tasks ***
Example task
    Set work item variable    username    MarkyMark
    Save Work Item

Python Example:

from RPA.Robocorp.WorkItems import WorkItems

customers = [{"id": 1, "name": "Apple"}, {"id": 2, "name": "Microsoft"}]
wi = WorkItems()
wi.get_input_work_item()
wi.set_work_item_variable("customers", customers)

set_work_item_variables(**kwargs)

Set multiple variables in the current work item.

Parameters

kwargs – Pairs of variable names and values

Example:

*** Tasks ***
Example task
    Set work item variables    username=MarkyMark    email=mark@example.com
    Save Work Item