RPA.Robocorp.Process

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

See https://robocorp.com/docs/control-room/operating-workforce for information about process run, step run and work item states.

See https://robocorp.com/docs/control-room/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()