Accessing (Cloud) UiPath Orchestrator through Python

Introduction

UiPath Orchestrator provides a platform for orchestrating robots, allowing users to manage and monitor their robotic process automation (RPA) deployments. This article presents a Python library that enables interaction with UiPath Orchestrator through its API. It covers authentication, retrieving assets, and handling responses. Additionally, note that the UiPath Orchestrator API offers more functions beyond retrieving assets.

Index

• Authentication
• Retrieving Assets

Authentication

To authenticate with UiPath Orchestrator, the library utilizes the provided credentials. The authentication process involves obtaining a token using the provided client ID and refresh token. Once authenticated, the token is stored for subsequent API requests.

Retrieving Assets

The library allows retrieving all assets stored in UiPath Orchestrator. It constructs the necessary API requests, including authentication headers and query parameters. The retrieved assets are then processed and returned as structured data. It's important to note that get_assets() is one of the examples of functions available in the library.

More Functions

Beyond retrieving assets, there are additional functions to interact with UiPath Orchestrator's API. These functions enable various operations such as managing robots, queues, processes, and more. Developers can explore the UiPath Orchestrator API documentation to discover the full range of functionalities available.

Code Examples

import base64
from dataclasses import dataclass
from datetime import datetime
import json
import logging
from pydantic import BaseModel, Field, TypeAdapter, field_validator
from requests.sessions import Session


@dataclass
class Configuration:
    url_base: str = None
    url_auth: str = None
    client_id: str = None
    refresh_token: str = None
    token: str = None
    unit_id: str = None


@dataclass
class Response:
    status_code: int
    content: list | dict | str | None = None


class UiPath(object):
    def __init__(self, credentials: dict) -> None:
        """
        Initializes variables and performs authentication.

        Args:
            credentials (dict): Credentials from Fortanix.
        """
        logging.basicConfig(level=logging.INFO)
        self.session: Session = Session()
        self.__configuration = self.auth(credentials=credentials)

    def __del__(self) -> None:
        """
        Clean the house at the exit.
        """
        logging.info(msg="Closes session")
        self.session.close()

    def is_auth(self) -> bool:
        """
        Checks whether authentication was successful.

        Returns:
            bool: If true, then authentication was successful.
        """
        logging.info(msg="Gets authentication status")
        return False if self.__configuration.token is None else True

    def auth(self, credentials: dict) -> Configuration:
        """
        Authenticate (On-Premise).
        Performs authentication process to obtain the token.

        Args:
            credentials (dict): Credentials from Fortanix.

        Returns:
            Configuration: Access configuration.
        """
        logging.info(msg="Opens session")
        configuration = Configuration()
        configuration.client_id = credentials["custom_metadata"]["client_id"]
        configuration.unit_id = credentials["custom_metadata"]["unit_id"]
        configuration.url_base = credentials["custom_metadata"]["url_base"]
        configuration.url_auth = credentials["custom_metadata"]["url_auth"]
        configuration.refresh_token = base64.b64decode(credentials["value"]).decode("utf-8")
        headers = {"Connection": "keep-alive", "Content-Type": "application/json"}
        body = {"grant_type": "refresh_token", "client_id": configuration.client_id,
                "refresh_token": configuration.refresh_token}
        response = self.session.post(url=configuration.url_auth, json=body, headers=headers)
        logging.info(msg=f"HTTP Status Code {response.status_code}")
        if response.status_code == 200:
            configuration.token = json.loads(response.content.decode("utf-8"))["access_token"]
            return configuration

    def get_assets(self, save_as: str | None = None) -> Response:
        """
        Assets - Get all.
        Gets the UiPath Orchestrator assets.

        Args:
            save_as (str, optional): Name of the Json file that contains the
              request response.

        Returns:
            Response (dataclass): Asset information and response status.
        """
        logging.info(msg="Gets a list of all assets")
        headers = {"Content-Type": "application/json", "Authorization": "Bearer " + self.__configuration.token,
                   "X-UIPATH-OrganizationUnitID": f"{self.__configuration.unit_id}"}
        url_query = f"{self.__configuration.url_base}odata/Assets"

        class DataStructure(BaseModel):
            Id: int = Field(alias="Id")
            Name: str = Field(alias="Name")
            ExternalName: str | None = Field(alias="ExternalName", default=None)
            HasDefaultValue: bool = Field(alias="HasDefaultValue")
            Value: str = Field(alias="Value")
            ValueScope: str = Field(alias="ValueScope")
            ValueType: str = Field(alias="ValueType")
            IntValue: int = Field(alias="IntValue")
            StringValue: str = Field(alias="StringValue")
            BoolValue: bool = Field(alias="BoolValue")
            CredentialUsername: str = Field(alias="CredentialUsername")
            CredentialStoreId: int | None = Field(alias="CredentialStoreId", default=None)
            CanBeDeleted: bool = Field(alias="CanBeDeleted")
            Description: str | None = Field(alias="Description", default=None)

        alias_list = [field.alias if field.alias is not None else field_name for field_name, field in
                      DataStructure.model_fields.items()]
        params = {"$select": ",".join(alias_list)}
        response = self.session.get(url=url_query, headers=headers, params=params, verify=True)
        logging.info(msg=f"HTTP Status Code {response.status_code}")
        content = None
        if response.status_code == 200:
            if save_as is not None:
                with open(file=save_as, mode="wb") as file:
                    file.write(response.content)
            content_raw = response.json()["value"]
            content = [dict(data) for data in TypeAdapter(list[DataStructure]).validate_python(content_raw)]

        return Response(status_code=response.status_code, content=content)

Conclusion

This Python library provides a convenient interface for accessing UiPath Orchestrator's API. By handling authentication and providing methods to retrieve assets, it simplifies integration with UiPath's automation platform. Developers can utilize this library to build custom automation solutions and manage their robotic processes efficiently. Additionally, the library offers more functions beyond retrieving assets, enabling a wide range of interactions with UiPath Orchestrator.