Python API

Process

class RPA.Robocorp.Process.Process(workspace_id: Optional[str] = None, process_id: Optional[str] = None, workspace_api_key: Optional[str] = None, **kwargs)

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
                )

ROBOT_AUTO_KEYWORDS = False

ROBOT_LIBRARY_DOC_FORMAT = 'REST'

ROBOT_LIBRARY_SCOPE = 'GLOBAL'

property base_api: str

create_input_work_item(payload: Optional[Any] = None, files: Optional[Union[str, List]] = 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_robot_run_artifact(process_run_id: str, step_run_id: str, artifact_id: str, filename: str, process_id: Optional[str] = None) → str

Get a download URL for a process run artifact

Parameters

• process_run_id – id of the process run

• step_run_id – id of the process step run

• artifact_id – id of the run artifact

• filename – filename of the run artifact

• process_id – specific process to which runs belongs to

Returns

url for file download

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: Dict[str, str]

list_process_run_work_items(process_run_id: Optional[str] = None, process_id: Optional[str] = None, include_data: bool = False, item_state: Optional[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

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

List Robot run artifacts

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

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: ConfigurationType = ConfigurationType.default, extra_info: Optional[Union[str, List]] = 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: Optional[Union[Dict, List[Dict]]] = 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