fix: remove typing optional

Author: cristipufu

src/uipath/runtime/base.py

@@ -5,7 +5,6 @@
     Any,
     AsyncGenerator,
     Literal,
-    Optional,
     Protocol,
 )
 
@@ -38,7 +37,7 @@ class UiPathExecuteOptions(BaseModel):
         default=False,
         description="Indicates whether to resume a suspended execution.",
     )
-    breakpoints: Optional[list[str] | Literal["*"]] = Field(
+    breakpoints: list[str] | Literal["*"] | None = Field(
         default=None,
         description="List of nodes or '*' to break on all steps.",
     )
@@ -57,8 +56,8 @@ class UiPathExecutableProtocol(Protocol):
 
     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:
         """Execute the runtime with the given input and options."""
         ...
@@ -69,8 +68,8 @@ class UiPathStreamableProtocol(Protocol):
 
     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]:
         """Stream execution events in real-time.
 
@@ -147,9 +146,9 @@ def __init__(
         delegate: UiPathRuntimeProtocol,
         trace_manager: UiPathTraceManager,
         root_span: str = "root",
-        span_attributes: Optional[dict[str, str]] = None,
-        log_handler: Optional[UiPathRuntimeExecutionLogHandler] = None,
-        execution_id: Optional[str] = None,
+        span_attributes: dict[str, str] | None = None,
+        log_handler: UiPathRuntimeExecutionLogHandler | None = None,
+        execution_id: str | None = None,
     ):
         """Initialize the executor."""
         self.delegate = delegate
@@ -163,8 +162,8 @@ def __init__(
 
     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:
         """Execute runtime with context."""
         if self.log_handler:
@@ -190,8 +189,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]:
         """Stream runtime execution with context.
 

src/uipath/runtime/context.py

@@ -6,11 +6,10 @@
 from functools import cached_property
 from typing import (
     Any,
-    Optional,
     TypeVar,
 )
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 from uipath.runtime.errors import (
     UiPathErrorCategory,
@@ -29,21 +28,21 @@
 class UiPathRuntimeContext(BaseModel):
     """Context information passed throughout the runtime execution."""
 
-    entrypoint: Optional[str] = None
-    input: Optional[dict[str, Any]] = None
-    job_id: Optional[str] = None
+    entrypoint: str | None = None
+    input: dict[str, Any] | None = None
+    job_id: str | None = None
     config_path: str = "uipath.json"
-    runtime_dir: Optional[str] = "__uipath"
+    runtime_dir: str | None = "__uipath"
     result_file: str = "output.json"
     state_file: str = "state.db"
-    input_file: Optional[str] = None
-    output_file: Optional[str] = None
-    trace_file: Optional[str] = None
-    logs_file: Optional[str] = "execution.log"
-    logs_min_level: Optional[str] = "INFO"
-    result: Optional[UiPathRuntimeResult] = None
+    input_file: str | None = None
+    output_file: str | None = None
+    trace_file: str | None = None
+    logs_file: str | None = "execution.log"
+    logs_min_level: str | None = "INFO"
+    result: UiPathRuntimeResult | None = None
 
-    model_config = {"arbitrary_types_allowed": True, "extra": "allow"}
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
 
     def __enter__(self):
         """Async enter method called when entering the 'async with' block.
@@ -188,7 +187,7 @@ def state_file_path(self) -> str:
         return os.path.join("__uipath", "state.db")
 
     @classmethod
-    def with_defaults(cls: type[C], config_path: Optional[str] = None, **kwargs) -> C:
+    def with_defaults(cls: type[C], config_path: str | None = None, **kwargs) -> C:
         """Construct a context with defaults, reading env vars and config file."""
         resolved_config_path = config_path or os.environ.get(
             "UIPATH_CONFIG_PATH", "uipath.json"
@@ -212,7 +211,7 @@ def with_defaults(cls: type[C], config_path: Optional[str] = None, **kwargs) ->
         return base
 
     @classmethod
-    def from_config(cls: type[C], config_path: Optional[str] = None, **kwargs) -> C:
+    def from_config(cls: type[C], config_path: str | None = None, **kwargs) -> C:
         """Load configuration from uipath.json file."""
         path = config_path or "uipath.json"
         config = {}

src/uipath/runtime/debug/runtime.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from typing import Any, AsyncGenerator, Optional, cast
+from typing import Any, AsyncGenerator, cast
 
 from uipath.runtime.base import (
     UiPathExecuteOptions,
@@ -56,8 +56,8 @@ def __init__(
 
     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:
         """Execute the workflow with debug support."""
         final_result = None
@@ -73,15 +73,15 @@ 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]:
         """Stream execution events with debug support."""
         try:
             await self.debug_bridge.connect()
             await self.debug_bridge.emit_execution_started()
 
-            result: Optional[UiPathRuntimeResult] = None
+            result: UiPathRuntimeResult | None = None
 
             # Try to stream events from inner runtime
             try:
@@ -107,8 +107,8 @@ async def stream(
 
     async def _stream_and_debug(
         self,
-        input: Optional[dict[str, Any]] = None,
-        options: Optional[UiPathExecuteOptions] = None,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         """Stream events from inner runtime and handle debug interactions."""
         final_result: UiPathRuntimeResult
@@ -185,7 +185,7 @@ async def _stream_and_debug(
                             )
                             await self.debug_bridge.emit_state_update(state_event)
 
-                            resume_data: Optional[dict[str, Any]] = None
+                            resume_data: dict[str, Any] | None = None
                             try:
                                 resume_data = await self._poll_trigger(
                                     final_result.trigger, self.delegate.trigger_manager
@@ -234,7 +234,7 @@ async def dispose(self) -> None:
 
     async def _poll_trigger(
         self, trigger: UiPathResumeTrigger, reader: UiPathResumeTriggerReaderProtocol
-    ) -> Optional[dict[str, Any]]:
+    ) -> dict[str, Any] | None:
         """Poll a resume trigger until data is available.
 
         Args:

src/uipath/runtime/errors/contract.py

@@ -1,7 +1,6 @@
 """Standard error contract used across the runtime."""
 
 from enum import Enum
-from typing import Optional
 
 from pydantic import BaseModel
 
@@ -38,6 +37,6 @@ class UiPathErrorContract(BaseModel):
     # - Deployment: Configuration, licensing, or permission issues
     # - System: Unexpected internal errors or infrastructure issues
 
-    status: Optional[int] = (
+    status: int | None = (
         None  # HTTP status code, if relevant (e.g., when forwarded from a web API)
     )

src/uipath/runtime/errors/exception.py

@@ -2,7 +2,7 @@
 
 import sys
 import traceback
-from typing import Any, Optional
+from typing import Any
 
 from uipath.runtime.errors.codes import UiPathErrorCode
 from uipath.runtime.errors.contract import UiPathErrorCategory, UiPathErrorContract
@@ -17,7 +17,7 @@ def __init__(
         title: str,
         detail: str,
         category: UiPathErrorCategory = UiPathErrorCategory.UNKNOWN,
-        status: Optional[int] = None,
+        status: int | None = None,
         prefix: str = "Python",
         include_traceback: bool = True,
     ):
@@ -42,13 +42,13 @@ def __init__(
         )
         super().__init__(detail)
 
-    def _extract_http_status(self) -> Optional[int]:
+    def _extract_http_status(self) -> int | None:
         """Extract HTTP status code from the exception chain if present."""
         exc_info = sys.exc_info()
         if not exc_info or len(exc_info) < 2 or exc_info[1] is None:
             return None
 
-        exc: Optional[BaseException] = exc_info[1]  # Current exception being handled
+        exc: BaseException | None = exc_info[1]  # Current exception being handled
         while exc is not None:
             if hasattr(exc, "status_code"):
                 return exc.status_code
@@ -85,7 +85,7 @@ def __init__(
         title: str,
         detail: str,
         category: UiPathErrorCategory = UiPathErrorCategory.UNKNOWN,
-        status: Optional[int] = None,
+        status: int | None = None,
         prefix: str = "Python",
         include_traceback: bool = True,
     ):

src/uipath/runtime/events/base.py

@@ -1,7 +1,7 @@
 """Base classes for UiPath runtime events."""
 
 from enum import Enum
-from typing import Any, Optional
+from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -21,10 +21,10 @@ class UiPathRuntimeEvent(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     event_type: UiPathRuntimeEventType
-    execution_id: Optional[str] = Field(
+    execution_id: str | None = Field(
         default=None, description="The runtime execution id associated with the event"
     )
-    metadata: Optional[dict[str, Any]] = Field(
+    metadata: dict[str, Any] | None = Field(
         default=None, description="Additional event context"
     )
 

src/uipath/runtime/events/state.py

@@ -1,6 +1,6 @@
 """Events related to agent messages and state updates."""
 
-from typing import Any, Optional
+from typing import Any
 
 from pydantic import Field
 
@@ -62,7 +62,7 @@ class UiPathRuntimeStateEvent(UiPathRuntimeEvent):
     """
 
     payload: dict[str, Any] = Field(description="Framework-specific state update")
-    node_name: Optional[str] = Field(
+    node_name: str | None = Field(
         default=None, description="Name of the node/agent that caused this update"
     )
     event_type: UiPathRuntimeEventType = Field(

src/uipath/runtime/logging/_context.py

@@ -1,9 +1,8 @@
 """Execution context tracking for logging."""
 
 from contextvars import ContextVar
-from typing import Optional
 
 # Context variable to track current execution_id
-current_execution_id: ContextVar[Optional[str]] = ContextVar(
+current_execution_id: ContextVar[str | None] = ContextVar(
     "current_execution_id", default=None
 )

src/uipath/runtime/logging/_interceptor.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import sys
-from typing import Optional, TextIO, Union, cast
+from typing import TextIO, cast
 
 from uipath.runtime.logging._context import current_execution_id
 from uipath.runtime.logging._filters import (
@@ -21,12 +21,12 @@ class UiPathRuntimeLogsInterceptor:
 
     def __init__(
         self,
-        min_level: Optional[str] = "INFO",
-        dir: Optional[str] = "__uipath",
-        file: Optional[str] = "execution.log",
-        job_id: Optional[str] = None,
-        execution_id: Optional[str] = None,
-        log_handler: Optional[logging.Handler] = None,
+        min_level: str | None = "INFO",
+        dir: str | None = "__uipath",
+        file: str | None = "execution.log",
+        job_id: str | None = None,
+        execution_id: str | None = None,
+        log_handler: logging.Handler | None = None,
     ):
         """Initialize the log interceptor.
 
@@ -56,11 +56,11 @@ def __init__(
         self.original_stdout = cast(TextIO, sys.stdout)
         self.original_stderr = cast(TextIO, sys.stderr)
 
-        self.log_handler: Union[
-            UiPathRuntimeFileLogsHandler,
-            logging.StreamHandler[TextIO],
-            logging.Handler,
-        ]
+        self.log_handler: (
+            UiPathRuntimeFileLogsHandler
+            | logging.StreamHandler[TextIO]
+            | logging.Handler
+        )
 
         if log_handler:
             self.log_handler = log_handler
@@ -82,7 +82,7 @@ def __init__(
         self.log_handler.setLevel(self.numeric_min_level)
 
         # Add execution context filter if execution_id provided
-        self.execution_filter: Optional[logging.Filter] = None
+        self.execution_filter: logging.Filter | None = None
         if execution_id:
             self.execution_filter = UiPathRuntimeExecutionFilter(execution_id)
             self.log_handler.addFilter(self.execution_filter)

src/uipath/runtime/result.py

@@ -1,7 +1,7 @@
 """Result of an execution with status and optional error information."""
 
 from enum import Enum
-from typing import Any, Optional, Union
+from typing import Any
 
 from pydantic import BaseModel, Field
 
@@ -21,10 +21,10 @@ class UiPathRuntimeStatus(str, Enum):
 class UiPathRuntimeResult(UiPathRuntimeEvent):
     """Result of an execution with status and optional error information."""
 
-    output: Optional[Union[dict[str, Any], BaseModel]] = None
+    output: dict[str, Any] | BaseModel | None = None
     status: UiPathRuntimeStatus = UiPathRuntimeStatus.SUCCESSFUL
-    trigger: Optional[UiPathResumeTrigger] = None
-    error: Optional[UiPathErrorContract] = None
+    trigger: UiPathResumeTrigger | None = None
+    error: UiPathErrorContract | None = None
 
     event_type: UiPathRuntimeEventType = Field(
         default=UiPathRuntimeEventType.RUNTIME_RESULT, frozen=True