Python API

Process

class RPA.Robocorp.Process.ConfigurationType

Bases: enum.Enum

Possible configuration types

default = 'default'
work_items = 'workItemIds'
class RPA.Robocorp.Process.Process(workspace_id: str = None, process_id: str = None, workspace_api_key: str = None, **kwargs)

Bases: object

A library for interacting with Control Room (CR) Process API endpoints.

See Operating Workforce for information about process run, step run and work item states.

See APIs and webhooks for information about Control Room APIs.

Examples

Robot Framework

In the following example a task creates two input work items, and starts a process with those items. This results in 2 different process runs in the Control Room.

*** Settings ***
Library    RPA.Robocorp.Process
Library    RPA.Robocorp.Vault

*** Keywords ***
Initialize Process Library
    ${secrets}=  Get Secret  ProcessAPI
    Set Credentials
    ...   ${secrets}[workspace_id]
    ...   ${secrets}[process_id]
    ...   ${secrets}[apikey]

*** Tasks ***
Start process with work items
    [Setup]   Initialize Process Library
    &{item1}=  Create Dictionary  fname=Mark  lname=Monkey
    &{item2}=  Create Dictionary  fname=John  lname=Doe
    @{items}=  Create List  ${item1}   ${item2}
    Start Process  work_items=${items}  batch=True

Robot Framework

In the following example a task creates work item with files. To include files in a work item, the item needs to be created before starting the process (note. different start keyword than above).

In this example I am using same keywords and settings from above example.

*** Tasks ***
Start process with work items
    [Setup]   Initialize Process Library
    &{data}=  Create Dictionary  fname=Mark  lname=Monkey
    @{files}=  Create List
    ...   ${CURDIR}${/}workdata.xlsx
    ...   ${CURDIR}${/}other.csv
    ${item_id}=    Create Input Work Item
    ...   payload=${data}
    ...   files=${files}
    Start Configured Process
    ...  config_type=work_items
    ...  extra_info=${item_id}

Python

List work items in Control Room and retry failed items.

from RPA.Robocorp.Process import Process
from RPA.Robocorp.Vault import Vault

secrets = Vault().get_secret("ProcessAPI")
process = Process(
    secrets["workspace_id"],
    secrets["process_id"],
    secrets["apikey"]
)


def retry_failed_items():
    items = process.list_process_work_items()
    for item in items:
        if item["state"] == "FAILED":
            print("FAILED work item: %s" % item["id"])
            result = process.retry_work_item(item["id"])
            print(result)

if __name__ == "__main__":
    retry_failed_items()
ROBOT_AUTO_KEYWORDS = False
ROBOT_LIBRARY_DOC_FORMAT = 'REST'
ROBOT_LIBRARY_SCOPE = 'GLOBAL'
property base_api
create_input_work_item(payload: Optional[Any] = None, files: Union[str, List, None] = None, process_id: Optional[str] = None) → int

Create an input work item for a process

Parameters
  • payload – work item data

  • files – absolute filepaths as single string or list

  • process_id – specific process to which item belongs to

Returns

The integer that represents the work item id

get_process_id_by_name(process_name: str, workspace_id: Optional[str] = None) → Optional[int]

Get a process id of the process by name

Parameters
  • process_name – name of the process in the Control Room

  • workspace_id – specific Control Room workspace to which process belongs to

Returns

the next iteration id or None

get_process_run_status(process_run_id: str, step_run_id: Optional[str] = None, process_id: Optional[str] = None) → Dict[str, Any]

Get a process run status by run id

Parameters
  • process_run_id – id of the process run

  • step_run_id – id of the process step run

  • process_id – specific process to which runs belongs to

Returns

the response JSON

get_work_item(workitem_id: str, include_data: bool = False, process_id: Optional[str] = None) → Dict[str, Any]

Get work item from Control Room

Parameters
  • workitem_id – id of the work item

  • include_data – include work item payload and files in the response (default False)

  • process_id – specific process to which runs belongs to

Returns

the JSON of the work items associated with a given process

property headers
list_process_run_work_items(process_run_id: str = None, process_id: str = None, include_data: bool = False, item_state: str = None)

List work items belonging to a specific process run

Parameters
  • process_run_id – specific process step run to which items belongs to

  • process_id – specific process to which items belongs to

  • include_data – include work item payload and files in the response (default False)

  • item_state – state of work items to return (default all)

list_process_runs(run_state: Optional[str] = None, limit: Optional[int] = 10, process_id: Optional[str] = None) → Union[str, None, Any]

List of runs related to a process

Parameters
  • run_state – state of runs to return (default all)

  • limit – number of runs to return (default 10)

  • process_id – specific process to which runs belongs to

Returns

the JSON data of the process runs based on the provided parameters

list_process_runs_in_workspace(run_state: Optional[str] = None, limit: Optional[int] = 10, workspace_id: Optional[str] = None) → Union[str, None, Any]

List all process runs in a workspace

Parameters
  • run_state – state of runs to return (default all)

  • limit – number of runs to return (default 10)

  • workspace_id – specific Control Room workspace to which process belongs to

Returns

the JSON data of the process runs based on the provided parameters

list_process_work_items(process_id: Optional[str] = None, include_data: bool = False, item_state: Optional[str] = None) → Union[str, None, Any]

List work items belonging to a process

Parameters
  • include_data – include work item payload and files in the response (default False)

  • item_state – state of work items to return (default all)

  • process_id – specific process to which items belongs to

Returns

the JSON data of the process runs based on the provided parameters

list_processes(workspace_id: Optional[str] = None) → Dict[str, Any]

List all processes in a workspace

Parameters

workspace_id – specific Control Room workspace to which process belongs to

Returns

the JSON data of the process runs based on the provided parameters

process_api(process_id: Optional[str] = None) → str
register_file_upload(filepath: str, workitem_filename: str, workitem_id: str, process_id: Optional[str] = None) → Tuple[str, Dict[str, Any]]
retry_work_item(work_item_id: str, process_id: Optional[str] = None) → Dict[str, Any]

Retry processing of work item in FAILED state

Parameters
  • work_item_id – ID of the work item to retry

  • process_id – specific process to start

Returns

the response JSON

set_apikey(apikey: Optional[str] = None) → None

Set Workspace API access key

Parameters

apikey – workspace API access key

set_credentials(workspace_id: Optional[str] = None, process_id: Optional[str] = None, apikey: Optional[str] = None) → None

Set credentials needed by the Process API

Parameters
  • workspace_id – ID of the Control Room workspace

  • process_id – ID of the Control Room process

  • apikey – workspace API access key

set_process_id(process_id: Optional[str] = None) → None

Set Control Room process ID

Parameters

process_id – ID of the Control Room process

set_workspace_id(workspace_id: Optional[str] = None) → None

Set Control Room workspace ID

Parameters

workspace_id – ID of the Control Room workspace

start_configured_process(config_type: RPA.Robocorp.Process.ConfigurationType = <ConfigurationType.default: 'default'>, extra_info: Union[str, List, None] = None, process_id: Optional[str] = None) → str

Start a Control Room process with the provided configuration

Parameters
  • config_type – type of the start, (ConfigurationType.default)

  • extra_info – data to be sent with the start, for example. work item IDs

  • process_id – specific process to start

Returns

string of the request response

start_process(work_items: Union[Dict, List[Dict], None] = None, batch: bool = False, process_id: Optional[str] = None) → Dict[str, Any]

Start a Control Room process

Parameters
  • work_items – input work items for the process (default empty)

  • batch – set to True if sending list of workitems to start each as a separate run

  • process_id – specific process to start

Returns

JSON of the request response

Table showing different results depending on parameter values.

work_items

batch

result

None

False

Trigger a process with empty a work item

None

True

Error. work_items needs to be a list

dict

False

Trigger a process with a work item containing payload of a dict

dict

True

Error. work_items needs to be a list

list

False

Trigger a process with a work item containing payload of a list

list

True

Trigger multiple process runs with work items each containing payload of a dict

upload_file_to_s3(filepath: str, workitem_filename: str, data: Any) → Tuple[str, str]
workspace_api(workspace_id: Optional[str] = None) → str
RPA.Robocorp.Process.to_configuration_type(value: Any) → RPA.Robocorp.Process.ConfigurationType

Convert value to ConfigurationType enum.