feat: extract common methods

Author: radu-mocanu

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.0.62"
+version = "2.0.63"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.10"
@@ -93,6 +93,7 @@ ignore-decorators = []
 "uipath/_cli/**" = ["D"]
 # TODO: Remove this once model documentation is added
 "src/uipath/models/**" = ["D101", "D100", "D104", "D102", "D107"]
+"src/uipath/runtime_commons/**" = ["D101", "D100", "D104", "D102", "D107", "D103"]
 
 [tool.ruff.format]
 quote-style = "double"

src/uipath/runtime_commons/__init__.py

@@ -0,0 +1,9 @@
+from .hitl import HitlProcessor, HitlReader
+from .tracing import setup_tracer_httpx_logging, simple_serialize_defaults
+
+__all__ = [
+    "HitlReader",
+    "HitlProcessor",
+    "setup_tracer_httpx_logging",
+    "simple_serialize_defaults",
+]

src/uipath/runtime_commons/hitl.py

@@ -0,0 +1,193 @@
+import json
+import uuid
+from dataclasses import dataclass
+from functools import cached_property
+from typing import Any, Optional
+
+from uipath import UiPath
+from uipath.models import CreateAction, InvokeProcess, WaitAction, WaitJob
+
+from .._cli._runtime._contracts import (
+    UiPathApiTrigger,
+    UiPathErrorCategory,
+    UiPathResumeTrigger,
+    UiPathResumeTriggerType,
+    UiPathRuntimeError,
+    UiPathRuntimeStatus,
+)
+
+
+def _try_convert_to_json_format(value: str) -> str:
+    try:
+        return json.loads(value)
+    except json.decoder.JSONDecodeError:
+        return value
+
+
+async def _get_api_payload(inbox_id: str) -> Any:
+    """Fetch payload data for API triggers.
+
+    Args:
+        inbox_id: The Id of the inbox to fetch the payload for.
+
+    Returns:
+        The value field from the API response payload, or None if an error occurs.
+    """
+    response = None
+    try:
+        uipath = UiPath()
+        response = uipath.api_client.request(
+            "GET",
+            f"/orchestrator_/api/JobTriggers/GetPayload/{inbox_id}",
+            include_folder_headers=True,
+        )
+        data = response.json()
+        return data.get("payload")
+    except Exception as e:
+        raise UiPathRuntimeError(
+            "API_CONNECTION_ERROR",
+            "Failed to get trigger payload",
+            f"Error fetching API trigger payload for inbox {inbox_id}: {str(e)}",
+            UiPathErrorCategory.SYSTEM,
+            response.status_code if response else None,
+        ) from e
+
+
+class HitlReader:
+    @classmethod
+    async def read(cls, resume_trigger: UiPathResumeTrigger) -> Optional[str]:
+        uipath = UiPath()
+        match resume_trigger.trigger_type:
+            case UiPathResumeTriggerType.ACTION:
+                if resume_trigger.item_key:
+                    action = await uipath.actions.retrieve_async(
+                        resume_trigger.item_key,
+                        app_folder_key=resume_trigger.folder_key,
+                        app_folder_path=resume_trigger.folder_path,
+                    )
+                    return action.data
+
+            case UiPathResumeTriggerType.JOB:
+                if resume_trigger.item_key:
+                    job = await uipath.jobs.retrieve_async(
+                        resume_trigger.item_key,
+                        folder_key=resume_trigger.folder_key,
+                        folder_path=resume_trigger.folder_path,
+                    )
+                    if (
+                        job.state
+                        and not job.state.lower()
+                        == UiPathRuntimeStatus.SUCCESSFUL.value.lower()
+                    ):
+                        raise UiPathRuntimeError(
+                            "INVOKED_PROCESS_FAILURE",
+                            "Invoked process did not finish successfully.",
+                            _try_convert_to_json_format(str(job.job_error or job.info)),
+                        )
+                    return job.output_arguments
+
+            case UiPathResumeTriggerType.API:
+                if resume_trigger.api_resume and resume_trigger.api_resume.inbox_id:
+                    return await _get_api_payload(resume_trigger.api_resume.inbox_id)
+
+            case _:
+                raise UiPathRuntimeError(
+                    "UNKNOWN_TRIGGER_TYPE",
+                    "Unexpected trigger type received",
+                    f"Trigger type :{type(resume_trigger.trigger_type)} is invalid",
+                    UiPathErrorCategory.USER,
+                )
+
+        raise UiPathRuntimeError(
+            "HITL_FEEDBACK_FAILURE",
+            "Failed to receive payload from HITL action",
+            detail="Failed to receive payload from HITL action",
+            category=UiPathErrorCategory.SYSTEM,
+        )
+
+
+@dataclass
+class HitlProcessor:
+    """Processes events in a Human-(Robot/Agent)-In-The-Loop scenario."""
+
+    value: Any
+
+    @cached_property
+    def type(self) -> UiPathResumeTriggerType:
+        """Returns the type of the interrupt value."""
+        if isinstance(self.value, CreateAction) or isinstance(self.value, WaitAction):
+            return UiPathResumeTriggerType.ACTION
+        if isinstance(self.value, InvokeProcess) or isinstance(self.value, WaitJob):
+            return UiPathResumeTriggerType.JOB
+        # default to API trigger
+        return UiPathResumeTriggerType.API
+
+    async def create_resume_trigger(self) -> Optional[UiPathResumeTrigger]:
+        """Returns the resume trigger."""
+        uipath = UiPath()
+        try:
+            hitl_input = self.value
+            resume_trigger = UiPathResumeTrigger(
+                trigger_type=self.type, payload=hitl_input.model_dump_json()
+            )
+            match self.type:
+                case UiPathResumeTriggerType.ACTION:
+                    resume_trigger.folder_path = hitl_input.app_folder_path
+                    resume_trigger.folder_key = hitl_input.app_folder_key
+                    if isinstance(hitl_input, WaitAction):
+                        resume_trigger.item_key = hitl_input.action.key
+                    elif isinstance(hitl_input, CreateAction):
+                        action = await uipath.actions.create_async(
+                            title=hitl_input.title,
+                            app_name=hitl_input.app_name if hitl_input.app_name else "",
+                            app_folder_path=hitl_input.app_folder_path
+                            if hitl_input.app_folder_path
+                            else "",
+                            app_folder_key=hitl_input.app_folder_key
+                            if hitl_input.app_folder_key
+                            else "",
+                            app_key=hitl_input.app_key if hitl_input.app_key else "",
+                            app_version=hitl_input.app_version
+                            if hitl_input.app_version
+                            else 1,
+                            assignee=hitl_input.assignee if hitl_input.assignee else "",
+                            data=hitl_input.data,
+                        )
+                        if action:
+                            resume_trigger.item_key = action.key
+
+                case UiPathResumeTriggerType.JOB:
+                    resume_trigger.folder_path = hitl_input.process_folder_path
+                    resume_trigger.folder_key = hitl_input.process_folder_key
+                    if isinstance(hitl_input, WaitJob):
+                        resume_trigger.item_key = hitl_input.job.key
+                    elif isinstance(hitl_input, InvokeProcess):
+                        job = await uipath.processes.invoke_async(
+                            name=hitl_input.name,
+                            input_arguments=hitl_input.input_arguments,
+                            folder_path=hitl_input.process_folder_path,
+                            folder_key=hitl_input.process_folder_key,
+                        )
+                        if job:
+                            resume_trigger.item_key = job.key
+
+                case UiPathResumeTriggerType.API:
+                    resume_trigger.api_resume = UiPathApiTrigger(
+                        inbox_id=str(uuid.uuid4()), request=hitl_input.prefix
+                    )
+                case _:
+                    raise UiPathRuntimeError(
+                        "UNKNOWN_HITL_MODEL",
+                        "Unexpected model received",
+                        f"{type(hitl_input)} is not a valid Human(Robot/Agent)-In-The-Loop model",
+                        UiPathErrorCategory.USER,
+                    )
+        except Exception as e:
+            raise UiPathRuntimeError(
+                "HITL_ACTION_CREATION_FAILED",
+                "Failed to create HITL action",
+                f"{str(e)}",
+                UiPathErrorCategory.SYSTEM,
+            ) from e
+
+        return resume_trigger

src/uipath/runtime_commons/tracing.py

@@ -0,0 +1,52 @@
+import datetime
+import logging
+from zoneinfo import ZoneInfo
+
+
+class IgnoreSpecificUrl(logging.Filter):
+    def __init__(self, url_to_ignore):
+        super().__init__()
+        self.url_to_ignore = url_to_ignore
+
+    def filter(self, record):
+        try:
+            if record.msg == 'HTTP Request: %s %s "%s %d %s"':
+                # Ignore the log if the URL matches the one we want to ignore
+                method = record.args[0]
+                url = record.args[1]
+
+                if method == "POST" and url.path.endswith(self.url_to_ignore):
+                    # Check if the URL contains the specific path we want to ignore
+                    return True
+                return False
+
+        except Exception:
+            return False
+
+
+def setup_tracer_httpx_logging(url: str):
+    # Create a custom logger for httpx
+    # Add the custom filter to the root logger
+    logging.getLogger("httpx").addFilter(IgnoreSpecificUrl(url))
+
+
+def simple_serialize_defaults(obj):
+    if hasattr(obj, "model_dump"):
+        return obj.model_dump(exclude_none=True, mode="json")
+    if hasattr(obj, "dict"):
+        return obj.dict()
+    if hasattr(obj, "to_dict"):
+        return obj.to_dict()
+
+    if isinstance(obj, (set, tuple)):
+        if hasattr(obj, "_asdict") and callable(obj._asdict):
+            return obj._asdict()
+        return list(obj)
+
+    if isinstance(obj, datetime.datetime):
+        return obj.isoformat()
+
+    if isinstance(obj, (datetime.timezone, ZoneInfo)):
+        return obj.tzname(None)
+
+    return str(obj)

src/uipath/runtime_commons/utils.py

@@ -0,0 +1,24 @@
+def serialize_object(obj):
+    """Recursively serializes an object and all its nested components."""
+    # Handle Pydantic models
+    if hasattr(obj, "model_dump"):
+        return serialize_object(obj.model_dump(by_alias=True))
+    elif hasattr(obj, "dict"):
+        return serialize_object(obj.dict())
+    elif hasattr(obj, "to_dict"):
+        return serialize_object(obj.to_dict())
+    # Handle dictionaries
+    elif isinstance(obj, dict):
+        return {k: serialize_object(v) for k, v in obj.items()}
+    # Handle lists
+    elif isinstance(obj, list):
+        return [serialize_object(item) for item in obj]
+    # Handle other iterable objects (convert to dict first)
+    elif hasattr(obj, "__iter__") and not isinstance(obj, (str, bytes)):
+        try:
+            return serialize_object(dict(obj))
+        except (TypeError, ValueError):
+            return obj
+    # Return primitive types as is
+    else:
+        return obj