Skip to content

refactor(RuntimeUpdates): decouple script runtime from a generic runtime #575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cristipufu merged 1 commit into from
Sep 15, 2025
Merged

refactor(RuntimeUpdates): decouple script runtime from a generic runtime #575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions src/uipath/_cli/_runtime/_contracts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,6 +473,21 @@ async def __aenter__(self):
with open(self.context.input_file) as f:
self.context.input = f.read()

try:
if self.context.input:
self.context.input_json = json.loads(self.context.input)
if self.context.input_json is None:
self.context.input_json = {}
except json.JSONDecodeError as e:
raise UiPathRuntimeError(
"INPUT_INVALID_JSON",
"Invalid JSON input",
f"The input data is not valid JSON: {str(e)}",
UiPathErrorCategory.USER,
) from e

await self.validate()

# Intercept all stdout/stderr/logs and write them to a file (runtime/evals), stdout (debug)
self.logs_interceptor = LogsInterceptor(
min_level=self.context.logs_min_level,
Expand Down
285 changes: 26 additions & 259 deletions src/uipath/_cli/_runtime/_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,30 +1,37 @@
"""Python script runtime implementation for executing and managing python scripts."""

import importlib.util
import inspect
import json
import logging
import os
from dataclasses import asdict, is_dataclass
from typing import Any, Dict, Optional, Type, TypeVar, cast, get_type_hints

from pydantic import BaseModel
from typing import Any, Awaitable, Callable, Optional, TypeVar

from ._contracts import (
UiPathBaseRuntime,
UiPathErrorCategory,
UiPathRuntimeContext,
UiPathRuntimeError,
UiPathRuntimeResult,
UiPathRuntimeStatus,
)
from ._script_executor import ScriptExecutor

logger = logging.getLogger(__name__)

T = TypeVar("T")
R = TypeVar("R")
AsyncFunc = Callable[[T], Awaitable[R]]


class UiPathRuntime(UiPathBaseRuntime):
"""Runtime for executing Python scripts."""
def __init__(self, context: UiPathRuntimeContext, executor: AsyncFunc[Any, Any]):
self.context = context
self.executor = executor

async def cleanup(self) -> None:
"""Cleanup runtime resources."""
pass

async def validate(self):
"""Validate runtime context."""
pass

async def execute(self) -> Optional[UiPathRuntimeResult]:
"""Execute the Python script with the provided input and configuration.
Expand All @@ -35,15 +42,8 @@ async def execute(self) -> Optional[UiPathRuntimeResult]:
Raises:
UiPathRuntimeError: If execution fails
"""
await self.validate()

try:
if self.context.entrypoint is None:
return None

script_result = await self._execute_python_script(
self.context.entrypoint, self.context.input_json
)
script_result = await self.executor(self.context.input_json)

if self.context.job_id is None:
logger.info(script_result)
Expand All @@ -65,248 +65,15 @@ async def execute(self) -> Optional[UiPathRuntimeResult]:
UiPathErrorCategory.SYSTEM,
) from e

async def validate(self) -> None:
"""Validate runtime inputs."""
if not self.context.entrypoint:
raise UiPathRuntimeError(
"ENTRYPOINT_MISSING",
"No entrypoint specified",
"Please provide a path to a Python script.",
UiPathErrorCategory.USER,
)

if not os.path.exists(self.context.entrypoint):
raise UiPathRuntimeError(
"ENTRYPOINT_NOT_FOUND",
"Script not found",
f"Script not found at path {self.context.entrypoint}.",
UiPathErrorCategory.USER,
)

try:
if self.context.input:
self.context.input_json = json.loads(self.context.input)
if self.context.input_json is None:
self.context.input_json = {}
except json.JSONDecodeError as e:
raise UiPathRuntimeError(
"INPUT_INVALID_JSON",
"Invalid JSON input",
f"The input data is not valid JSON: {str(e)}",
UiPathErrorCategory.USER,
) from e

async def cleanup(self) -> None:
"""Cleanup runtime resources."""
pass

async def _execute_python_script(self, script_path: str, input_data: Any) -> Any:
"""Execute the Python script with the given input."""
spec = importlib.util.spec_from_file_location("dynamic_module", script_path)
if not spec or not spec.loader:
raise UiPathRuntimeError(
"IMPORT_ERROR",
"Module import failed",
f"Could not load spec for {script_path}",
UiPathErrorCategory.USER,
)

module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except Exception as e:
raise UiPathRuntimeError(
"MODULE_EXECUTION_ERROR",
"Module execution failed",
f"Error executing module: {str(e)}",
UiPathErrorCategory.USER,
) from e

for func_name in ["main", "run", "execute"]:
if hasattr(module, func_name):
main_func = getattr(module, func_name)
sig = inspect.signature(main_func)
params = list(sig.parameters.values())

# Check if the function is asynchronous
is_async = inspect.iscoroutinefunction(main_func)

# Case 1: No parameters
if not params:
try:
result = await main_func() if is_async else main_func()
return (
self._convert_from_class(result)
if result is not None
else {}
)
except Exception as e:
raise UiPathRuntimeError(
"FUNCTION_EXECUTION_ERROR",
f"Error executing {func_name} function",
f"Error: {str(e)}",
UiPathErrorCategory.USER,
) from e

input_param = params[0]
input_type = input_param.annotation

# Case 2: Class, dataclass, or Pydantic model parameter
if input_type != inspect.Parameter.empty and (
is_dataclass(input_type)
or self._is_pydantic_model(input_type)
or hasattr(input_type, "__annotations__")
):
try:
valid_type = cast(Type[Any], input_type)
typed_input = self._convert_to_class(input_data, valid_type)
result = (
await main_func(typed_input)
if is_async
else main_func(typed_input)
)
return (
self._convert_from_class(result)
if result is not None
else {}
)
except Exception as e:
raise UiPathRuntimeError(
"FUNCTION_EXECUTION_ERROR",
f"Error executing {func_name} function with typed input",
f"Error: {str(e)}",
UiPathErrorCategory.USER,
) from e

# Case 3: Dict parameter
else:
try:
result = (
await main_func(input_data)
if is_async
else main_func(input_data)
)
return (
self._convert_from_class(result)
if result is not None
else {}
)
except Exception as e:
raise UiPathRuntimeError(
"FUNCTION_EXECUTION_ERROR",
f"Error executing {func_name} function with dictionary input",
f"Error: {str(e)}",
UiPathErrorCategory.USER,
) from e

raise UiPathRuntimeError(
"ENTRYPOINT_FUNCTION_MISSING",
"No entry function found",
f"No main function (main, run, or execute) found in {script_path}",
UiPathErrorCategory.USER,
)

def _convert_to_class(self, data: Dict[str, Any], cls: Type[T]) -> T:
"""Convert a dictionary to either a dataclass, Pydantic model, or regular class instance."""
# Handle Pydantic models
try:
if inspect.isclass(cls) and issubclass(cls, BaseModel):
return cls.model_validate(data)
except TypeError:
# issubclass can raise TypeError if cls is not a class
pass

# Handle dataclasses
if is_dataclass(cls):
field_types = get_type_hints(cls)
converted_data = {}

for field_name, field_type in field_types.items():
if field_name not in data:
continue

value = data[field_name]
if (
is_dataclass(field_type)
or self._is_pydantic_model(field_type)
or hasattr(field_type, "__annotations__")
) and isinstance(value, dict):
typed_field = cast(Type[Any], field_type)
value = self._convert_to_class(value, typed_field)
converted_data[field_name] = value

return cls(**converted_data)

# Handle regular classes
else:
sig = inspect.signature(cls.__init__)
params = sig.parameters

init_args = {}

for param_name, param in params.items():
if param_name == "self":
continue

if param_name in data:
value = data[param_name]
param_type = (
param.annotation
if param.annotation != inspect.Parameter.empty
else Any
)

if (
is_dataclass(param_type)
or self._is_pydantic_model(param_type)
or hasattr(param_type, "__annotations__")
) and isinstance(value, dict):
typed_param = cast(Type[Any], param_type)
value = self._convert_to_class(value, typed_param)

init_args[param_name] = value
elif param.default != inspect.Parameter.empty:
init_args[param_name] = param.default

return cls(**init_args)

def _is_pydantic_model(self, cls: Type[Any]) -> bool:
"""Safely check if a class is a Pydantic model."""
try:
return inspect.isclass(cls) and issubclass(cls, BaseModel)
except TypeError:
# issubclass can raise TypeError if cls is not a class
return False

def _convert_from_class(self, obj: Any) -> Dict[str, Any]:
"""Convert a class instance (dataclass, Pydantic model, or regular) to a dictionary."""
if obj is None:
return {}

# Handle Pydantic models
if isinstance(obj, BaseModel):
return obj.model_dump()
class UiPathScriptRuntime(UiPathRuntime):
"""Runtime for executing Python scripts."""

# Handle dataclasses
elif is_dataclass(obj):
# Make sure obj is an instance, not a class
if isinstance(obj, type):
return {}
return asdict(obj)
def __init__(self, context: UiPathRuntimeContext, entrypoint: str):
executor = ScriptExecutor(entrypoint)
super().__init__(context, executor)

# Handle regular classes
elif hasattr(obj, "__dict__"):
result = {}
for key, value in obj.__dict__.items():
# Skip private attributes
if not key.startswith("_"):
if (
isinstance(value, BaseModel)
or hasattr(value, "__dict__")
or is_dataclass(value)
):
result[key] = self._convert_from_class(value)
else:
result[key] = value
return result
return {} if obj is None else {str(type(obj).__name__): str(obj)} # Fallback
@classmethod
def from_context(cls, context: UiPathRuntimeContext):
"""Create runtime instance from context."""
return UiPathScriptRuntime(context, context.entrypoint or "")
Loading
Loading