Python API

class RPA.Robocorp.WorkItems.BaseAdapter

Bases: abc.ABC

Abstract base class for work item adapters.

abstract add_file(item_id: str, name: str, *, original_name: str, content: bytes)

Attach file to work item.

abstract create_output(parent_id: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None] = None) → str

Create new output for work item, and return created ID.

abstract get_file(item_id: str, name: str) → bytes

Read file’s contents from work item.

abstract list_files(item_id: str) → List[str]

List attached files in work item.

abstract load_payload(item_id: str) → Union[Dict[str, Any], List[Any], str, int, float, bool, None]

Load JSON payload from work item.

abstract release_input(item_id: str, state: RPA.Robocorp.WorkItems.State, exception: Optional[dict] = None)

Release the lastly retrieved input work item and set state.

abstract remove_file(item_id: str, name: str)

Remove attached file from work item.

abstract reserve_input() → str

Get next work item ID from the input queue and reserve it.

abstract save_payload(item_id: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None])

Save JSON payload to work item.

exception RPA.Robocorp.WorkItems.EmptyQueue

Bases: IndexError

Raised when trying to load an input item and none available.

args

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class RPA.Robocorp.WorkItems.Error

Bases: enum.Enum

Failed work item error type.

APPLICATION = 'APPLICATION'

BUSINESS = 'BUSINESS'

class RPA.Robocorp.WorkItems.FileAdapter(*args, **kwargs)

Bases: RPA.Robocorp.WorkItems.BaseAdapter

Adapter for mocking work item input queues.

Reads inputs from the given database file, and writes all created output items into an adjacent file with the suffix <filename>.output.json. If the output path is provided by an env var explicitly, then the file will be saved with the provided path and name.

Reads and writes all work item files from/to the same parent folder as the given input database.

Optional environment variables:

• RPA_INPUT_WORKITEM_PATH: Path to work items input database file

• RPA_OUTPUT_WORKITEM_PATH: Path to work items output database file

add_file(item_id: str, name: str, *, original_name: str, content: bytes)

Attach file to work item.

create_output(_: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None] = None) → str

Create new output for work item, and return created ID.

get_file(item_id: str, name: str) → bytes

Read file’s contents from work item.

property input_path

list_files(item_id: str) → List[str]

List attached files in work item.

load_database() → List

load_payload(item_id: str) → Union[Dict[str, Any], List[Any], str, int, float, bool, None]

Load JSON payload from work item.

property output_path

release_input(item_id: str, state: RPA.Robocorp.WorkItems.State, exception: Optional[dict] = None)

Release the lastly retrieved input work item and set state.

remove_file(item_id: str, name: str)

Remove attached file from work item.

reserve_input() → str

Get next work item ID from the input queue and reserve it.

save_payload(item_id: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None])

Save JSON payload to work item.

class RPA.Robocorp.WorkItems.RobocorpAdapter(*args, **kwargs)

Bases: RPA.Robocorp.WorkItems.BaseAdapter

Adapter for saving/loading work items from Robocorp Control Room.

Required environment variables:

• RC_API_WORKITEM_HOST: Work item API hostname

• RC_API_WORKITEM_TOKEN: Work item API access token

• RC_API_PROCESS_HOST: Process API hostname

• RC_API_PROCESS_TOKEN: Process API access token

• RC_WORKSPACE_ID: Control room workspace ID

• RC_PROCESS_ID: Control room process ID

• RC_PROCESS_RUN_ID: Control room process run ID

• RC_ROBOT_RUN_ID: Control room robot run ID

• RC_WORKITEM_ID: Control room work item ID (input)

add_file(item_id: str, name: str, *, original_name: str, content: bytes)

Attach file to work item.

create_output(parent_id: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None] = None) → str

Create new output for work item, and return created ID.

file_id(item_id: str, name: str) → str

get_file(item_id: str, name: str) → bytes

Read file’s contents from work item.

list_files(item_id: str) → List[str]

List attached files in work item.

load_payload(item_id: str) → Union[Dict[str, Any], List[Any], str, int, float, bool, None]

Load JSON payload from work item.

release_input(item_id: str, state: RPA.Robocorp.WorkItems.State, exception: Optional[dict] = None)

Release the lastly retrieved input work item and set state.

remove_file(item_id: str, name: str)

Remove attached file from work item.

reserve_input() → str

Get next work item ID from the input queue and reserve it.

save_payload(item_id: str, payload: Union[Dict[str, Any], List[Any], str, int, float, bool, None])

Save JSON payload to work item.

class RPA.Robocorp.WorkItems.State

Bases: enum.Enum

Work item state. (set when released)

DONE = 'COMPLETED'

FAILED = 'FAILED'

class RPA.Robocorp.WorkItems.WorkItem(adapter, item_id=None, parent_id=None)

Bases: object

Base class for input and output work items.

Parameters

• adapter – Adapter instance

• item_id – Work item ID (optional)

• parent_id – Parent work item’s ID (optional)

adapter = None

Adapter for loading/saving content

add_file(path, name=None)

Add file to current work item. Does not upload until save() is called.

Parameters

• path – Path to file to upload

• name – Name of file in work item. If not given, name of file on disk is used.

property files

List of filenames, including local files pending upload and excluding files pending removal.

get_file(name, path=None)

Load an attached file and store it on the local filesystem.

Parameters

• name – Name of attached file

• path – Destination path. Default to current working directory.

Returns

Path to created file

id: Optional[str] = None

This item’s and/or parent’s ID

property is_dirty

Check if work item has unsaved changes.

load()

Load data payload and list of files.

property payload

remove_file(name, missing_ok=True)

Remove file from current work item. Change is not applied until save() is called.

Parameters

name – Name of attached file

save()

Save data payload and attach/remove files.

state: Optional[State] = None

Item’s state on release; can be set once

class RPA.Robocorp.WorkItems.WorkItems(autoload: bool = True, root: Optional[str] = None, default_adapter: Union[Type[RPA.Robocorp.WorkItems.BaseAdapter], str] = <class 'RPA.Robocorp.WorkItems.RobocorpAdapter'>, auto_parse_email: Optional[Dict[str, str]] = {'rawEmail': 'parsedEmail'})

Bases: object

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.

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.

Development and mocking

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.

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)

ROBOT_AUTO_KEYWORDS = False

ROBOT_LIBRARY_DOC_FORMAT = 'REST'

ROBOT_LIBRARY_SCOPE = 'GLOBAL'

ROBOT_LISTENER_API_VERSION = 2

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:

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:

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:

Clear work item
Save work item

create_output_work_item()

Create a new output work item.

An output work item is always created as a child for an input item, and as such requires an input to be loaded.

All changes to the work item are done locally, and are only sent to the output queue after the keyword Save work item is called.

Example:

${customers}=    Load customer data
FOR    ${customer}    IN    @{customers}
    Create output work item
    Set work item variables    name=${customer.name}    id=${customer.id}
    Save work item
END

property current

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:

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:

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() → RPA.Robocorp.WorkItems.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:

${input} =    Get Current Work Item
${output} =   Create Output Work Item
Set Current Work Item    ${input}

get_input_work_item(_internal_call: bool = False)

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)

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:

${path}=    Get work item file    input.xls
Open workbook    ${path}

get_work_item_files(pattern, dirname=None)

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:

${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:

${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

Example:

${username}=    Get work item variable    username    default=guest

get_work_item_variables()

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

Example:

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

inputs: List[WorkItem] = None

Input work items

list_work_item_files()

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

Example:

${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:

${variables}=    List work item variables
Log    Available variables in work item: ${variables}

outputs: List[WorkItem] = None

Output work items

release_input_work_item(state: RPA.Robocorp.WorkItems.State, _auto_release: bool = False, exception_type: Optional[RPA.Robocorp.WorkItems.Error] = None, code: Optional[str] = None, message: Optional[str] = None)

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:

Login into portal
    ${user} =     Get Work Item Variable    user
    Log     Logging in ${user}
    Release Input Work Item     FAILED      exception_type=BUSINESS   code=LOGIN_PORTAL_DOWN     message=Unable to login into the portal – not proceeding  # noqa

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:

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:

Remove work item files    *.xlsx
Save work item

root = None

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: RPA.Robocorp.WorkItems.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.

Example:

${input}=    Get input work item
${output}=   Create output work item
Set current work item    ${input}

set_task_variables_from_work_item()

Convert all variables in the current work item to Robot Framework task variables.

Example:

# 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.

Example:

${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

Example:

Set work item variable    username    MarkyMark
Save work item

set_work_item_variables(**kwargs)

Set multiple variables in the current work item.

Parameters

kwargs – Pairs of variable names and values

Example:

Set work item variables    username=MarkyMark    email=mark@example.com
Save work item