feat: add per-service error extraction to EnrichedException (#1455)

Author: andreitava-uip

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.0.26"
+version = "0.0.27"
 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/errors/__init__.py

@@ -16,7 +16,7 @@
 
 from ._base_url_missing_error import BaseUrlMissingError
 from ._batch_transform_not_complete_exception import BatchTransformNotCompleteException
-from ._enriched_exception import EnrichedException
+from ._enriched_exception import EnrichedException, ExtractedErrorInfo
 from ._folder_not_found_exception import FolderNotFoundException
 from ._ingestion_in_progress_exception import IngestionInProgressException
 from ._operation_failed_exception import OperationFailedException
@@ -28,10 +28,11 @@
     "BaseUrlMissingError",
     "BatchTransformNotCompleteException",
     "EnrichedException",
+    "ExtractedErrorInfo",
     "FolderNotFoundException",
     "IngestionInProgressException",
-    "SecretMissingError",
-    "OperationNotCompleteException",
     "OperationFailedException",
+    "OperationNotCompleteException",
+    "SecretMissingError",
     "UnsupportedDataSourceException",
 ]

packages/uipath-platform/src/uipath/platform/errors/_enriched_exception.py

@@ -1,29 +1,58 @@
+from dataclasses import dataclass
+from functools import cached_property
+
 from httpx import HTTPStatusError
 
 
+@dataclass(frozen=True)
+class ExtractedErrorInfo:
+    message: str | None = None
+    error_code: str | None = None
+    trace_id: str | None = None
+
+
 class EnrichedException(Exception):
     """Enriched HTTP error with detailed request/response information.
 
-    This exception wraps HTTPStatusError and provides additional context about
-    the failed HTTP request, including URL, method, status code, and response content.
+    Wraps HTTPStatusError with URL, method, status code, and truncated response
+    content in __str__. For structured error fields, use ``error_info`` which
+    delegates to per-service extractors.
+
+    Does not retain a reference to the original HTTPStatusError — all needed
+    data is eagerly extracted. Callers needing the raw response can still
+    access ``__cause__`` (set by ``raise EnrichedException(e) from e``).
     """
 
     def __init__(self, error: HTTPStatusError) -> None:
-        # Extract the relevant details from the HTTPStatusError
-        self.status_code = error.response.status_code if error.response else "Unknown"
-        self.url = str(error.request.url) if error.request else "Unknown"
-        self.http_method = (
+        # while status code 0 is the correct one according to http standard;
+        # it has a totally oposite meaning as return codes in CLIs;
+        # opted for -1 to avoid confusion
+        self.status_code: int = error.response.status_code if error.response else -1
+        self.url: str = str(error.request.url) if error.request else "Unknown"
+        self.http_method: str = (
             error.request.method
             if error.request and error.request.method
             else "Unknown"
         )
+
+        self._response_body: str | None = None
+        self._content_type: str | None = None
+        if error.response is not None:
+            self._content_type = error.response.headers.get("content-type")
+            if error.response.content:
+                try:
+                    self._response_body = error.response.content.decode("utf-8")
+                except Exception:
+                    pass
+
         max_content_length = 200
-        if error.response and error.response.content:
-            content = error.response.content.decode("utf-8")
-            if len(content) > max_content_length:
-                self.response_content = content[:max_content_length] + "... (truncated)"
+        if self._response_body:
+            if len(self._response_body) > max_content_length:
+                self.response_content = (
+                    self._response_body[:max_content_length] + "... (truncated)"
+                )
             else:
-                self.response_content = content
+                self.response_content = self._response_body
         else:
             self.response_content = "No content"
 
@@ -34,5 +63,13 @@ def __init__(self, error: HTTPStatusError) -> None:
             f"\nResponse Content: {self.response_content}"
         )
 
-        # Initialize the parent Exception class with the formatted message
         super().__init__(enriched_message)
+
+    @cached_property
+    def error_info(self) -> ExtractedErrorInfo | None:
+        """Service-aware extraction of message, error_code, trace_id."""
+        from ._extractors._router import extract_error_info
+
+        if not self._response_body:
+            return None
+        return extract_error_info(self.url, self._response_body, self._content_type)

packages/uipath-platform/src/uipath/platform/errors/_extractors/_agenthub.py

@@ -0,0 +1,24 @@
+"""AgentHub / AgentsRuntime error payload extractor."""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_typed_field
+
+
+def extract_agenthub(body: dict[str, Any]) -> ExtractedErrorInfo:
+    message = get_typed_field(body, str, "message", "title")
+    error_code = get_typed_field(body, str, "errorCode", "code")
+    trace_id = get_typed_field(body, str, "traceId")
+
+    problem = get_typed_field(body, dict, "problem")
+    if problem is not None:
+        message = message or get_typed_field(problem, str, "detail", "title")
+        error_code = error_code or get_typed_field(problem, str, "errorCode", "type")
+        trace_id = trace_id or get_typed_field(problem, str, "traceId")
+
+    return ExtractedErrorInfo(
+        message=message,
+        error_code=error_code,
+        trace_id=trace_id,
+    )

packages/uipath-platform/src/uipath/platform/errors/_extractors/_apps.py

@@ -0,0 +1,14 @@
+"""Apps service error payload extractor."""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_typed_field
+
+
+def extract_apps(body: dict[str, Any]) -> ExtractedErrorInfo:
+    return ExtractedErrorInfo(
+        message=get_typed_field(body, str, "errorMessageV2", "message"),
+        error_code=get_typed_field(body, str, "errorCode", "code"),
+        trace_id=get_typed_field(body, str, "traceId"),
+    )

packages/uipath-platform/src/uipath/platform/errors/_extractors/_elements.py

@@ -0,0 +1,20 @@
+"""Elements service error payload extractor."""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_str_field, get_typed_field
+
+
+def extract_elements(body: dict[str, Any]) -> ExtractedErrorInfo:
+    message = get_typed_field(body, str, "providerMessage") or get_typed_field(
+        body, str, "message"
+    )
+    error_code = get_str_field(body, "providerErrorCode", "status")
+    trace_id = get_typed_field(body, str, "requestId")
+
+    return ExtractedErrorInfo(
+        message=message,
+        error_code=error_code,
+        trace_id=trace_id,
+    )

packages/uipath-platform/src/uipath/platform/errors/_extractors/_generic.py

@@ -0,0 +1,35 @@
+"""Generic fallback error payload extractor.
+
+Tries structural RFC 7807 detection first, then common field patterns
+covering Orchestrator, Connections, ECS, and most other services.
+"""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_field, get_str_field, get_typed_field
+from ._rfc7807 import extract_rfc7807, looks_like_rfc7807
+
+
+def extract_generic(body: dict[str, Any]) -> ExtractedErrorInfo:
+    if looks_like_rfc7807(body):
+        return extract_rfc7807(body)
+
+    message: str | None
+    error = get_field(body, "error")
+    if isinstance(error, str):
+        message = error
+    elif isinstance(error, dict):
+        message = get_typed_field(error, str, "message")
+    else:
+        message = get_typed_field(body, str, "message")
+
+    # str() conversion handles services that return numeric error codes
+    error_code = get_str_field(body, "errorCode", "code")
+    trace_id = get_typed_field(body, str, "traceId", "requestId")
+
+    return ExtractedErrorInfo(
+        message=message,
+        error_code=error_code,
+        trace_id=trace_id,
+    )

packages/uipath-platform/src/uipath/platform/errors/_extractors/_helpers.py

@@ -0,0 +1,63 @@
+"""Shared helpers for error payload extractors."""
+
+from typing import Any, TypeVar
+from urllib.parse import urlparse
+
+T = TypeVar("T")
+
+
+def get_field(body: dict[str, Any], *keys: str) -> Any:
+    """Return the first non-None value found for the given keys.
+
+    Automatically tries both camelCase and PascalCase for each key.
+    """
+    for key in keys:
+        val = body.get(key)
+        if val is not None:
+            return val
+        val = body.get(key[0].swapcase() + key[1:])
+        if val is not None:
+            return val
+    return None
+
+
+def get_typed_field(body: dict[str, Any], type_: type[T], *keys: str) -> T | None:
+    """Return the first non-None value matching *type_* for the given keys.
+
+    Skips values that don't match the expected type, trying the next key.
+    """
+    for key in keys:
+        val = body.get(key)
+        if val is None:
+            val = body.get(key[0].swapcase() + key[1:])
+        if isinstance(val, type_):
+            return val
+    return None
+
+
+def get_str_field(body: dict[str, Any], *keys: str) -> str | None:
+    """Return the first scalar value for the given keys, converted to str.
+
+    Skips dicts, lists, and None.
+    """
+    for key in keys:
+        val = body.get(key)
+        if val is None:
+            val = body.get(key[0].swapcase() + key[1:])
+        if val is not None and not isinstance(val, (dict, list)):
+            return str(val)
+    return None
+
+
+def extract_service_prefix(url: str) -> str | None:
+    """Extract the service prefix (e.g. 'orchestrator_') from a UiPath URL path."""
+    path = urlparse(url).path if "://" in url else url
+    for segment in path.strip("/").split("/"):
+        if segment.endswith("_") and len(segment) > 1:
+            return segment
+    return None
+
+
+def is_llm_path(url: str) -> bool:
+    """Check if the URL targets the LLM Gateway."""
+    return "/llm/" in url

packages/uipath-platform/src/uipath/platform/errors/_extractors/_llm_gateway.py

@@ -0,0 +1,23 @@
+"""LLM Gateway error payload extractor."""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_typed_field
+
+
+def extract_llm_gateway(body: dict[str, Any]) -> ExtractedErrorInfo:
+    message = get_typed_field(body, str, "message")
+    error_code = get_typed_field(body, str, "errorCode", "code")
+    trace_id = get_typed_field(body, str, "traceId")
+
+    error = get_typed_field(body, dict, "error")
+    if error is not None:
+        message = message or get_typed_field(error, str, "message")
+        error_code = error_code or get_typed_field(error, str, "code", "type")
+
+    return ExtractedErrorInfo(
+        message=message,
+        error_code=error_code,
+        trace_id=trace_id,
+    )

packages/uipath-platform/src/uipath/platform/errors/_extractors/_llmops.py

@@ -0,0 +1,14 @@
+"""LLMOps service error payload extractor."""
+
+from typing import Any
+
+from .._enriched_exception import ExtractedErrorInfo
+from ._helpers import get_typed_field
+
+
+def extract_llmops(body: dict[str, Any]) -> ExtractedErrorInfo:
+    return ExtractedErrorInfo(
+        message=get_typed_field(body, str, "errorMessage"),
+        error_code=get_typed_field(body, str, "code"),
+        trace_id=get_typed_field(body, str, "requestId"),
+    )