A library for interacting with Control Room (CR) Process API endpoints.
See Unattended processes for information about process run, step run and work
item states.
See APIs and webhooks for information about Control Room APIs.
The Process API endpoint is defined by RC_API_PROCESS_HOST environment variable,
which is available during Robocorp Workforce Agent runs.
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
)