RPA.Robocorp.Process

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}

Download from process runs artifacts all matching files

*** Settings ***
Library      RPA.Robocorp.Process
Library      RPA.Robocorp.Vault
Library      RPA.HTTP
Task Setup   Set Control Room Variables

*** Keywords ***
Download Artifacts Matching
    [Arguments]   ${filematch}
    @{workitems}=   List Process Work Items
    FOR  ${item}  IN  @{workitems}
        @{artifacts}=   List Run Artifacts
        ...  process_run_id=${item}[processRunId]
        ...  step_run_id=${item}[activityRunId]
        FOR  ${artifact}  IN  @{artifacts}
            IF  "${filematch}"  IN   "${artifact}[fileName]"
                ${download_link}=   Get Robot Run Artifact
                ...  process_run_id=${item}[processRunId]
                ...  step_run_id=${item}[activityRunId]
                ...  artifact_id=${artifact}[id]
                ...  filename=${artifact}[fileName]
                Download
                ...  url=${download_link}
                ...  target=%{ROBOT_ARTIFACTS}${/}${artifact}[fileName]
                ...  overwrite=${TRUE}
                ...  stream=${TRUE}
            END
        END
    END

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()

Download from process runs artifacts all ".xlsx" files

from RPA.Robocorp.Process import Process
from RPA.HTTP import HTTP

def download_artifacts_matching(filematch=".xlsx"):
    work_items = process.list_process_work_items()
    for item in work_items:
        artifacts = process.list_run_artifacts(
            process_run_id=item["processRunId"],
            step_run_id=item["activityRunId"]
        )
        for artifact in artifacts:
            if filematch in artifact["fileName"]:
                download_link = process.get_robot_run_artifact(
                    process_run_id=item["processRunId"],
                    step_run_id=item["activityRunId"],
                    artifact_id=artifact["id"],
                    filename=artifact["fileName"]
                )
                target_filepath = os.path.join(
                    os.getenv("ROBOT_ARTIFACTS"),
                    f"{artifact['fileName']}"
                )
                HTTP().download(
                    url=download_link,
                    target_file=target_filepath,
                    overwrite=True,
                    stream=True
                )