DEV Community

Cover image for Accessing (Cloud) UiPath Orchestrator through Python
≀Paulo Portela
≀Paulo Portela

Posted on

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)
Enter fullscreen mode Exit fullscreen mode

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.

API Trace View

How I Cut 22.3 Seconds Off an API Call with Sentry 🕒

Struggling with slow API calls? Dan Mindru walks through how he used Sentry's new Trace View feature to shave off 22.3 seconds from an API call.

Get a practical walkthrough of how to identify bottlenecks, split tasks into multiple parallel tasks, identify slow AI model calls, and more.

Read more →

Top comments (0)

Image of Docusign

🛠️ Bring your solution into Docusign. Reach over 1.6M customers.

Docusign is now extensible. Overcome challenges with disconnected products and inaccessible data by bringing your solutions into Docusign and publishing to 1.6M customers in the App Center.

Learn more