RPA.Robocorp.Process
module RPA.Process
class RPA.Robocorp.Process.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.
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.
Download from process runs artifacts all matching files
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
)
variable ROBOT_AUTO_KEYWORDS
ROBOT_AUTO_KEYWORDS = False
variable ROBOT_LIBRARY_DOC_FORMAT
ROBOT_LIBRARY_DOC_FORMAT = 'REST'
variable ROBOT_LIBRARY_SCOPE
ROBOT_LIBRARY_SCOPE = 'GLOBAL'
property base_api : str
property base_api : str
method create_input_work_item
create_input_work_item(payload: Optional[Any] = None, files: Optional[Union[str, List]] = None, process_id: Optional[str] = None)
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
method get_process_id_by_name
get_process_id_by_name(process_name: str, workspace_id: Optional[str] = None)
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
method get_process_run_status
get_process_run_status(process_run_id: str, step_run_id: Optional[str] = None, process_id: Optional[str] = None)
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
method get_robot_run_artifact
get_robot_run_artifact(process_run_id: str, step_run_id: str, artifact_id: str, filename: str, process_id: Optional[str] = None)
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
method get_work_item
get_work_item(workitem_id: str, include_data: bool = False, process_id: Optional[str] = None)
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]
property headers : Dict[str, str]
method list_process_run_work_items
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)
method list_process_runs
list_process_runs(run_state: Optional[str] = None, limit: Optional[int] = 10, process_id: Optional[str] = None)
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
method list_process_runs_in_workspace
list_process_runs_in_workspace(run_state: Optional[str] = None, limit: Optional[int] = 10, workspace_id: Optional[str] = None)
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
method list_process_work_items
list_process_work_items(process_id: Optional[str] = None, include_data: bool = False, item_state: Optional[str] = None)
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
method list_processes
list_processes(workspace_id: Optional[str] = None)
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
method list_run_artifacts
list_run_artifacts(process_run_id: str, step_run_id: str, process_id: Optional[str] = None)
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
method process_api
process_api(process_id: Optional[str] = None)
method register_file_upload
register_file_upload(filepath: str, workitem_filename: str, workitem_id: str, process_id: Optional[str] = None)
method retry_work_item
retry_work_item(work_item_id: str, process_id: Optional[str] = None)
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
method set_apikey
set_apikey(apikey: Optional[str] = None)
Set Workspace API access key
- Parameters: apikey – workspace API access key
method set_credentials
set_credentials(workspace_id: Optional[str] = None, process_id: Optional[str] = None, apikey: Optional[str] = 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
method set_process_id
set_process_id(process_id: Optional[str] = None)
Set Control Room process ID
- Parameters: process_id – ID of the Control Room process
method set_workspace_id
set_workspace_id(workspace_id: Optional[str] = None)
Set Control Room workspace ID
- Parameters: workspace_id – ID of the Control Room workspace
method start_configured_process
start_configured_process(config_type: ConfigurationType = ConfigurationType.default, extra_info: Optional[Union[str, List]] = None, process_id: Optional[str] = None)
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
method start_process
start_process(work_items: Optional[Union[Dict, List[Dict]]] = None, batch: bool = False, process_id: Optional[str] = None)
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 |
method upload_file_to_s3
upload_file_to_s3(filepath: str, workitem_filename: str, data: Any)
method workspace_api
workspace_api(workspace_id: Optional[str] = None)