refactor: move auth services in platform

Author: radugheo

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.0.23"
+version = "0.0.24"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/common/_external_application_service.py

@@ -3,10 +3,10 @@
 from urllib.parse import urlparse
 
 import httpx
-from httpx import HTTPStatusError, Request
+from httpx import HTTPStatusError
 
 from ..errors import EnrichedException
-from ._http_config import get_httpx_client_kwargs
+from ..identity import IdentityService
 from .auth import TokenData
 from .constants import ENV_BASE_URL
 
@@ -22,16 +22,6 @@ def __init__(self, base_url: Optional[str]):
         self._base_url = resolved_base_url
         self._domain = self._extract_environment_from_base_url(self._base_url)
 
-    def get_token_url(self) -> str:
-        """Get the token URL for the specified domain."""
-        match self._domain:
-            case "alpha":
-                return "https://alpha.uipath.com/identity_/connect/token"
-            case "staging":
-                return "https://staging.uipath.com/identity_/connect/token"
-            case _:  # cloud (default)
-                return "https://cloud.uipath.com/identity_/connect/token"
-
     def _is_valid_domain_or_subdomain(self, hostname: str, domain: str) -> bool:
         """Check if hostname is either an exact match or a valid subdomain of the domain.
 
@@ -87,53 +77,45 @@ def get_token_data(
         Returns:
             Token data if successful
         """
-        token_url = self.get_token_url()
-
-        data = {
-            "grant_type": "client_credentials",
-            "client_id": client_id,
-            "client_secret": client_secret,
-            "scope": scope,
+        domain_map = {
+            "alpha": "https://alpha.uipath.com",
+            "staging": "https://staging.uipath.com",
         }
+        domain = domain_map.get(self._domain, "https://cloud.uipath.com")
 
         try:
-            with httpx.Client(**get_httpx_client_kwargs()) as client:
-                response = client.post(token_url, data=data)
-                match response.status_code:
-                    case 200:
-                        return TokenData.model_validate(response.json())
-                    case 400:
-                        raise EnrichedException(
-                            HTTPStatusError(
-                                message="Invalid client credentials or request parameters.",
-                                request=Request(
-                                    data=data, url=token_url, method="post"
-                                ),
-                                response=response,
-                            )
+            return IdentityService.get_client_credentials_token(
+                domain=domain,
+                client_id=client_id,
+                client_secret=client_secret,
+                scope=scope,
+            )
+        except HTTPStatusError as e:
+            match e.response.status_code:
+                case 400:
+                    raise EnrichedException(
+                        HTTPStatusError(
+                            message="Invalid client credentials or request parameters.",
+                            request=e.request,
+                            response=e.response,
                         )
-                    case 401:
-                        raise EnrichedException(
-                            HTTPStatusError(
-                                message="Unauthorized: Invalid client credentials.",
-                                request=Request(
-                                    data=data, url=token_url, method="post"
-                                ),
-                                response=response,
-                            )
+                    ) from e
+                case 401:
+                    raise EnrichedException(
+                        HTTPStatusError(
+                            message="Unauthorized: Invalid client credentials.",
+                            request=e.request,
+                            response=e.response,
                         )
-                    case _:
-                        raise EnrichedException(
-                            HTTPStatusError(
-                                message=f"Authentication failed with unexpected status: {response.status_code}",
-                                request=Request(
-                                    data=data, url=token_url, method="post"
-                                ),
-                                response=response,
-                            )
+                    ) from e
+                case _:
+                    raise EnrichedException(
+                        HTTPStatusError(
+                            message=f"Authentication failed with unexpected status: {e.response.status_code}",
+                            request=e.request,
+                            response=e.response,
                         )
-        except EnrichedException:
-            raise
+                    ) from e
         except httpx.RequestError as e:
             raise Exception(f"Network error during authentication: {e}") from e
         except Exception as e:

packages/uipath-platform/src/uipath/platform/identity/__init__.py

@@ -0,0 +1,5 @@
+"""UiPath Identity Service."""
+
+from ._identity_service import IdentityService
+
+__all__ = ["IdentityService"]

packages/uipath-platform/src/uipath/platform/identity/_identity_service.py

@@ -0,0 +1,80 @@
+"""Identity service for UiPath authentication token operations."""
+
+from typing import Optional
+
+import httpx
+
+from ..common._http_config import get_httpx_client_kwargs
+from ..common.auth import TokenData
+
+
+class IdentityService:
+    """Service for interacting with the UiPath Identity server."""
+
+    @staticmethod
+    def refresh_access_token(
+        domain: str,
+        refresh_token: str,
+        client_id: str,
+    ) -> TokenData:
+        """Refresh an access token using a refresh token.
+
+        Args:
+            domain: The base URL of the UiPath identity server (e.g., "https://cloud.uipath.com").
+            refresh_token: The refresh token to exchange for a new access token.
+            client_id: The client ID of the application.
+
+        Returns:
+            TokenData containing the new access token and related information.
+
+        Raises:
+            httpx.HTTPStatusError: If the server returns a non-2xx response.
+            httpx.ConnectError: If there is a network connectivity issue.
+        """
+        url = f"{domain}/identity_/connect/token"
+        data = {
+            "grant_type": "refresh_token",
+            "refresh_token": refresh_token,
+            "client_id": client_id,
+        }
+
+        with httpx.Client(**get_httpx_client_kwargs()) as client:
+            response = client.post(url, data=data)
+            response.raise_for_status()
+            return TokenData.model_validate(response.json())
+
+    @staticmethod
+    def get_client_credentials_token(
+        domain: str,
+        client_id: str,
+        client_secret: str,
+        scope: Optional[str] = "OR.Execution",
+    ) -> TokenData:
+        """Obtain an access token using client credentials grant.
+
+        Args:
+            domain: The base URL of the UiPath identity server (e.g., "https://cloud.uipath.com").
+            client_id: The client ID of the application.
+            client_secret: The client secret of the application.
+            scope: The requested OAuth scopes (optional, default: "OR.Execution").
+
+        Returns:
+            TokenData containing the access token and related information.
+
+        Raises:
+            httpx.HTTPStatusError: If the server returns a non-2xx response.
+            httpx.ConnectError: If there is a network connectivity issue.
+        """
+        scope = scope or "OR.Execution"
+        url = f"{domain}/identity_/connect/token"
+        data = {
+            "grant_type": "client_credentials",
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "scope": scope,
+        }
+
+        with httpx.Client(**get_httpx_client_kwargs()) as client:
+            response = client.post(url, data=data)
+            response.raise_for_status()
+            return TokenData.model_validate(response.json())

packages/uipath-platform/src/uipath/platform/orchestrator/__init__.py

@@ -11,6 +11,7 @@
 from ._mcp_service import McpService
 from ._processes_service import ProcessesService
 from ._queues_service import QueuesService
+from ._studio_web_service import StudioWebService
 from .assets import Asset, UserAsset
 from .attachment import Attachment
 from .buckets import Bucket, BucketFile
@@ -34,6 +35,7 @@
     "McpService",
     "ProcessesService",
     "QueuesService",
+    "StudioWebService",
     "Asset",
     "UserAsset",
     "Attachment",

packages/uipath-platform/src/uipath/platform/orchestrator/_studio_web_service.py

@@ -0,0 +1,69 @@
+"""StudioWeb service for UiPath Platform."""
+
+import logging
+
+import httpx
+
+from ..common._http_config import get_httpx_client_kwargs
+
+logger = logging.getLogger(__name__)
+
+
+class StudioWebService:
+    """Service for interacting with the UiPath StudioWeb API."""
+
+    @staticmethod
+    def enable_first_run(tenant_url: str, access_token: str) -> None:
+        """Fire-and-forget POST requests to enable first run for StudioWeb.
+
+        Posts to TryEnableFirstRun and AcquireLicense endpoints.
+
+        Args:
+            tenant_url: The tenant base URL (e.g., "https://cloud.uipath.com/org/tenant").
+            access_token: The Bearer access token for authorization.
+        """
+        urls = [
+            f"{tenant_url}/orchestrator_/api/StudioWeb/TryEnableFirstRun",
+            f"{tenant_url}/orchestrator_/api/StudioWeb/AcquireLicense",
+        ]
+        headers = {"Authorization": f"Bearer {access_token}"}
+
+        with httpx.Client(**get_httpx_client_kwargs()) as client:
+            for url in urls:
+                try:
+                    response = client.post(url, headers=headers)
+                    if not response.is_success:
+                        logger.warning(
+                            "StudioWeb enable_first_run: POST %s returned %s",
+                            url,
+                            response.status_code,
+                        )
+                except httpx.HTTPError as exc:
+                    logger.warning(
+                        "StudioWeb enable_first_run: POST %s failed: %s",
+                        url,
+                        exc,
+                    )
+
+    @staticmethod
+    def get_server_version(domain: str) -> str | None:
+        """Get the Orchestrator server version.
+
+        Args:
+            domain: The base URL of the UiPath platform (e.g., "https://cloud.uipath.com").
+
+        Returns:
+            The server version string, or None if the request fails for any reason.
+        """
+        url = f"{domain}/orchestrator_/api/status/version"
+
+        try:
+            client_kwargs = get_httpx_client_kwargs()
+            client_kwargs["timeout"] = 5.0
+            with httpx.Client(**client_kwargs) as client:
+                response = client.get(url)
+                response.raise_for_status()
+                data = response.json()
+                return data.get("version")
+        except Exception:
+            return None

packages/uipath-platform/src/uipath/platform/portal/__init__.py

@@ -0,0 +1,11 @@
+"""UiPath Portal Service."""
+
+from ._portal_service import PortalService
+from .portal import OrganizationInfo, TenantInfo, TenantsAndOrganizationInfoResponse
+
+__all__ = [
+    "PortalService",
+    "TenantInfo",
+    "OrganizationInfo",
+    "TenantsAndOrganizationInfoResponse",
+]

packages/uipath-platform/src/uipath/platform/portal/_portal_service.py

@@ -0,0 +1,38 @@
+"""Portal service for UiPath Platform."""
+
+import httpx
+
+from ..common._http_config import get_httpx_client_kwargs
+from .portal import TenantsAndOrganizationInfoResponse
+
+
+class PortalService:
+    """Service for interacting with the UiPath Portal API."""
+
+    @staticmethod
+    def get_tenants_and_organizations(
+        domain: str,
+        prt_id: str,
+        access_token: str,
+    ) -> TenantsAndOrganizationInfoResponse:
+        """Retrieve tenants and organization info for the given organization.
+
+        Args:
+            domain: The base URL of the UiPath platform (e.g., "https://cloud.uipath.com").
+            prt_id: The organization/partition ID used in the URL path.
+            access_token: The Bearer access token for authorization.
+
+        Returns:
+            TenantsAndOrganizationInfoResponse containing tenants and organization info.
+
+        Raises:
+            httpx.HTTPStatusError: If the server returns a non-2xx response.
+            httpx.ConnectError: If there is a network connectivity issue.
+        """
+        url = f"{domain}/{prt_id}/portal_/api/filtering/leftnav/tenantsAndOrganizationInfo"
+        headers = {"Authorization": f"Bearer {access_token}"}
+
+        with httpx.Client(**get_httpx_client_kwargs()) as client:
+            response = client.get(url, headers=headers)
+            response.raise_for_status()
+            return TenantsAndOrganizationInfoResponse.model_validate(response.json())

packages/uipath-platform/src/uipath/platform/portal/portal.py

@@ -0,0 +1,24 @@
+"""Models for UiPath Portal service."""
+
+from pydantic import BaseModel
+
+
+class TenantInfo(BaseModel):
+    """Model representing a tenant."""
+
+    name: str
+    id: str
+
+
+class OrganizationInfo(BaseModel):
+    """Model representing an organization."""
+
+    id: str
+    name: str
+
+
+class TenantsAndOrganizationInfoResponse(BaseModel):
+    """Model representing the tenants and organization info response."""
+
+    tenants: list[TenantInfo]
+    organization: OrganizationInfo