fix: remove typing optional union

Author: cristipufu

src/uipath/dev/_demo/mock_context_runtime.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Any, AsyncGenerator, Optional
+from typing import Any, AsyncGenerator
 
 from opentelemetry import trace
 from uipath.runtime import (
@@ -57,8 +57,8 @@ async def get_schema(self) -> UiPathRuntimeSchema:
 
     async def execute(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathExecuteOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
     ) -> UiPathRuntimeResult:
         from uipath.runtime.debug import UiPathBreakpointResult
 
@@ -451,8 +451,8 @@ async def execute(
 
     async def stream(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathStreamOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         logger.info("MockRuntime: stream() invoked")
         print("[MockRuntime] stream() invoked")

src/uipath/dev/_demo/mock_greeting_runtime.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Any, AsyncGenerator, Optional
+from typing import Any, AsyncGenerator
 
 from opentelemetry import trace
 from uipath.runtime import (
@@ -59,8 +59,8 @@ async def get_schema(self) -> UiPathRuntimeSchema:
 
     async def execute(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathExecuteOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
     ) -> UiPathRuntimeResult:
         payload = input or {}
         name = str(payload.get("name", "world")).strip() or "world"
@@ -121,8 +121,8 @@ async def execute(
 
     async def stream(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathStreamOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         logger.info("GreetingRuntime: stream() invoked")
         yield await self.execute(input=input, options=options)

src/uipath/dev/_demo/mock_numbers_runtime.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Any, AsyncGenerator, Optional
+from typing import Any, AsyncGenerator
 
 from opentelemetry import trace
 from uipath.runtime import (
@@ -58,8 +58,8 @@ async def get_schema(self) -> UiPathRuntimeSchema:
 
     async def execute(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathExecuteOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
     ) -> UiPathRuntimeResult:
         payload = input or {}
         numbers = payload.get("numbers") or []
@@ -134,8 +134,8 @@ async def execute(
 
     async def stream(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathStreamOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         logger.info("NumberAnalyticsRuntime: stream() invoked")
         yield await self.execute(input=input, options=options)

src/uipath/dev/_demo/mock_support_runtime.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Any, AsyncGenerator, Optional
+from typing import Any, AsyncGenerator
 
 from opentelemetry import trace
 from uipath.runtime import (
@@ -57,8 +57,8 @@ async def get_schema(self) -> UiPathRuntimeSchema:
 
     async def execute(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathExecuteOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
     ) -> UiPathRuntimeResult:
         payload = input or {}
         message = str(payload.get("message", "")).strip()
@@ -136,8 +136,8 @@ async def execute(
 
     async def stream(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathStreamOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         logger.info("SupportChatRuntime: stream() invoked")
         yield await self.execute(input=input, options=options)

src/uipath/dev/models/execution.py

@@ -2,7 +2,7 @@
 
 import os
 from datetime import datetime
-from typing import Any, Optional, Union
+from typing import Any
 from uuid import uuid4
 
 from rich.text import Text
@@ -17,7 +17,7 @@ class ExecutionRun:
     def __init__(
         self,
         entrypoint: str,
-        input_data: Union[dict[str, Any]],
+        input_data: dict[str, Any],
         conversational: bool = False,
         debug: bool = False,
     ):
@@ -27,14 +27,14 @@ def __init__(
         self.input_data = input_data
         self.conversational = conversational
         self.debug = debug
-        self.resume_data: Optional[dict[str, Any]] = None
-        self.output_data: Optional[dict[str, Any]] = None
+        self.resume_data: dict[str, Any] | None = None
+        self.output_data: dict[str, Any] | None = None
         self.start_time = datetime.now()
-        self.end_time: Optional[datetime] = None
+        self.end_time: datetime | None = None
         self.status = "pending"  # pending, running, completed, failed, suspended
         self.traces: list[TraceMessage] = []
         self.logs: list[LogMessage] = []
-        self.error: Optional[UiPathErrorContract] = None
+        self.error: UiPathErrorContract | None = None
 
     @property
     def duration(self) -> str:

src/uipath/dev/models/messages.py

@@ -1,7 +1,7 @@
 """Messages used for inter-component communication in the UiPath Developer Console."""
 
 from datetime import datetime
-from typing import Any, Optional, Union
+from typing import Any
 
 from rich.console import RenderableType
 from textual.message import Message
@@ -14,8 +14,8 @@ def __init__(
         self,
         run_id: str,
         level: str,
-        message: Union[str, RenderableType],
-        timestamp: Optional[datetime] = None,
+        message: str | RenderableType,
+        timestamp: datetime | None = None,
     ):
         """Initialize a LogMessage instance."""
         self.run_id = run_id
@@ -33,12 +33,12 @@ def __init__(
         run_id: str,
         span_name: str,
         span_id: str,
-        parent_span_id: Optional[str] = None,
-        trace_id: Optional[str] = None,
+        parent_span_id: str | None = None,
+        trace_id: str | None = None,
         status: str = "running",
-        duration_ms: Optional[float] = None,
-        timestamp: Optional[datetime] = None,
-        attributes: Optional[dict[str, Any]] = None,
+        duration_ms: float | None = None,
+        timestamp: datetime | None = None,
+        attributes: dict[str, Any] | None = None,
     ):
         """Initialize a TraceMessage instance."""
         self.run_id = run_id

src/uipath/dev/services/debug_bridge.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from typing import Any, Callable, Literal, Optional
+from typing import Any, Callable, Literal
 
 from uipath.runtime.debug import UiPathBreakpointResult, UiPathDebugQuitError
 from uipath.runtime.events import UiPathRuntimeStateEvent
@@ -22,15 +22,11 @@ def __init__(self):
         self._breakpoints: list[str] | Literal["*"] = "*"  # Default: step mode
 
         # Callbacks to UI
-        self.on_execution_started: Optional[Callable[[], None]] = None
-        self.on_state_update: Optional[Callable[[UiPathRuntimeStateEvent], None]] = None
-        self.on_breakpoint_hit: Optional[Callable[[UiPathBreakpointResult], None]] = (
-            None
-        )
-        self.on_execution_completed: Optional[Callable[[UiPathRuntimeResult], None]] = (
-            None
-        )
-        self.on_execution_error: Optional[Callable[[str], None]] = None
+        self.on_execution_started: Callable[[], None] | None = None
+        self.on_state_update: Callable[[UiPathRuntimeStateEvent], None] | None = None
+        self.on_breakpoint_hit: Callable[[UiPathBreakpointResult], None] | None = None
+        self.on_execution_completed: Callable[[UiPathRuntimeResult], None] | None = None
+        self.on_execution_error: Callable[[str], None] | None = None
 
     async def connect(self) -> None:
         """Establish connection to debugger."""

src/uipath/dev/services/run_service.py

@@ -4,7 +4,7 @@
 
 import traceback
 from datetime import datetime
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable
 
 from pydantic import BaseModel
 from uipath.core.tracing import UiPathTraceManager
@@ -40,14 +40,14 @@ def __init__(
         self,
         runtime_factory: UiPathRuntimeFactoryProtocol,
         trace_manager: UiPathTraceManager,
-        on_run_updated: Optional[RunUpdatedCallback] = None,
-        on_log: Optional[LogCallback] = None,
-        on_trace: Optional[TraceCallback] = None,
+        on_run_updated: RunUpdatedCallback | None = None,
+        on_log: LogCallback | None = None,
+        on_trace: TraceCallback | None = None,
     ) -> None:
         """Initialize RunService with runtime factory and trace manager."""
         self.runtime_factory = runtime_factory
         self.trace_manager = trace_manager
-        self.runs: Dict[str, ExecutionRun] = {}
+        self.runs: dict[str, ExecutionRun] = {}
 
         self.on_run_updated = on_run_updated
         self.on_log = on_log
@@ -68,7 +68,7 @@ def register_run(self, run: ExecutionRun) -> None:
         self.runs[run.id] = run
         self._emit_run_updated(run)
 
-    def get_run(self, run_id: str) -> Optional[ExecutionRun]:
+    def get_run(self, run_id: str) -> ExecutionRun | None:
         """Get a registered run."""
         return self.runs.get(run_id)
 
@@ -78,7 +78,7 @@ async def execute(self, run: ExecutionRun) -> None:
         This is the extracted version of the old `_execute_runtime` method.
         """
         try:
-            execution_input: Optional[dict[str, Any]] = {}
+            execution_input: dict[str, Any] | None = {}
             execution_options: UiPathExecuteOptions = UiPathExecuteOptions()
 
             if run.status == "suspended":
@@ -239,7 +239,7 @@ def handle_trace(self, trace_msg: TraceMessage) -> None:
         if self.on_trace is not None:
             self.on_trace(trace_msg)
 
-    def get_debug_bridge(self, run_id: str) -> Optional[TextualDebugBridge]:
+    def get_debug_bridge(self, run_id: str) -> TextualDebugBridge | None:
         """Get the debug bridge for a run."""
         return self.debug_bridges.get(run_id)
 

src/uipath/dev/ui/panels/new_run_panel.py

@@ -1,7 +1,7 @@
 """Panel for creating new runs with entrypoint selection and JSON input."""
 
 import json
-from typing import Any, Dict, Tuple, cast
+from typing import Any, Tuple, cast
 
 from textual.app import ComposeResult
 from textual.containers import Container, Horizontal, Vertical
@@ -96,7 +96,7 @@ def __init__(
 
         self.entrypoints: list[str] = []
 
-        self.entrypoint_schemas: Dict[str, dict[str, Any]] = {}
+        self.entrypoint_schemas: dict[str, dict[str, Any]] = {}
 
         self.conversational: bool = False
         self.initial_input: str = "{}"

src/uipath/dev/ui/panels/run_details_panel.py

@@ -1,7 +1,5 @@
 """Panel for displaying execution run details, traces, and logs."""
 
-from typing import Optional
-
 from textual.app import ComposeResult
 from textual.containers import Container, Horizontal, Vertical
 from textual.reactive import reactive
@@ -82,7 +80,7 @@ def show_span_details(self, trace_msg: TraceMessage):
 class RunDetailsPanel(Container):
     """Panel showing traces and logs for selected run with tabbed interface."""
 
-    current_run: reactive[Optional[ExecutionRun]] = reactive(None)
+    current_run: reactive[ExecutionRun | None] = reactive(None)
 
     def __init__(self, **kwargs):
         """Initialize RunDetailsPanel."""
@@ -145,7 +143,7 @@ def compose(self) -> ComposeResult:
                 )
 
     def watch_current_run(
-        self, old_value: Optional[ExecutionRun], new_value: Optional[ExecutionRun]
+        self, old_value: ExecutionRun | None, new_value: ExecutionRun | None
     ):
         """Watch for changes to the current run."""
         if new_value is not None: