refactor(RemoveABC): favoring protocols

Allow ducktyping!

Author: akshaylive

src/uipath/runtime/__init__.py

@@ -1,15 +1,15 @@
 """UiPath Runtime Package."""
 
 from uipath.runtime.base import (
-    UiPathBaseRuntime,
     UiPathExecuteOptions,
     UiPathExecutionRuntime,
+    UiPathRuntimeProtocol,
     UiPathStreamNotSupportedError,
     UiPathStreamOptions,
 )
 from uipath.runtime.context import UiPathRuntimeContext
 from uipath.runtime.events import UiPathRuntimeEvent
-from uipath.runtime.factory import UiPathRuntimeFactory
+from uipath.runtime.factory import UiPathRuntimeCreator, UiPathRuntimeScanner
 from uipath.runtime.result import (
     UiPathApiTrigger,
     UiPathBreakpointResult,
@@ -23,9 +23,10 @@
     "UiPathExecuteOptions",
     "UiPathStreamOptions",
     "UiPathRuntimeContext",
-    "UiPathBaseRuntime",
+    "UiPathRuntimeProtocol",
     "UiPathExecutionRuntime",
-    "UiPathRuntimeFactory",
+    "UiPathRuntimeCreator",
+    "UiPathRuntimeScanner",
     "UiPathRuntimeResult",
     "UiPathRuntimeStatus",
     "UiPathRuntimeEvent",

src/uipath/runtime/base.py

@@ -1,18 +1,15 @@
 """Base runtime class and async context manager implementation."""
 
 import logging
-from abc import ABC, abstractmethod
+import typing
 from typing import (
     Any,
     AsyncGenerator,
-    Generic,
     Literal,
     Optional,
-    TypeVar,
 )
 
 from pydantic import BaseModel, Field
-from typing_extensions import override
 from uipath.core import UiPathTraceManager
 
 from uipath.runtime.events import (
@@ -55,27 +52,20 @@ class UiPathStreamOptions(UiPathExecuteOptions):
     pass
 
 
-class UiPathBaseRuntime(ABC):
-    """Base runtime class implementing the async context manager protocol.
+class UiPathExecutable(typing.Protocol):
+    """UiPath execution interface."""
 
-    This allows using the class with 'async with' statements.
-    """
-
-    async def get_schema(self) -> UiPathRuntimeSchema:
-        """Get schema for this runtime.
-
-        Returns: The runtime's schema (entrypoint type, input/output json schema).
-        """
-        raise NotImplementedError()
-
-    @abstractmethod
     async def execute(
         self,
         input: Optional[dict[str, Any]] = None,
         options: Optional[UiPathExecuteOptions] = None,
     ) -> UiPathRuntimeResult:
         """Produce the agent output."""
-        raise NotImplementedError()
+        ...
+
+
+class UiPathStreamable(typing.Protocol):
+    """UiPath streaming interface."""
 
     async def stream(
         self,
@@ -120,21 +110,35 @@ async def stream(
         # Without it, the function wouldn't match the AsyncGenerator return type
         yield
 
-    @abstractmethod
-    async def cleanup(self):
-        """Cleaup runtime resources."""
-        pass
 
+class HasSchema(typing.Protocol):
+    """Contains runtime input and output schema."""
 
-T = TypeVar("T", bound=UiPathBaseRuntime)
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        """Get schema for a runtime.
+
+        Returns: The runtime's schema (entrypoint type, input/output json schema).
+        """
+        ...
 
 
-class UiPathExecutionRuntime(UiPathBaseRuntime, Generic[T]):
+# Note: explicitly marking it as a protocol for mypy.
+# https://mypy.readthedocs.io/en/stable/protocols.html#defining-subprotocols-and-subclassing-protocols
+# Note that inheriting from an existing protocol does not automatically turn the subclass into a protocol
+# – it just creates a regular (non-protocol) class or ABC that implements the given protocol (or protocols).
+# The Protocol base class must always be explicitly present if you are defining a protocol.
+class UiPathRuntimeProtocol(
+    UiPathExecutable, UiPathStreamable, HasSchema, typing.Protocol
+):
+    """UiPath Runtime Protocol."""
+
+
+class UiPathExecutionRuntime:
     """Handles runtime execution with tracing/telemetry."""
 
     def __init__(
         self,
-        delegate: T,
+        delegate: UiPathRuntimeProtocol,
         trace_manager: UiPathTraceManager,
         root_span: str = "root",
         log_handler: Optional[UiPathRuntimeExecutionLogHandler] = None,
@@ -174,7 +178,6 @@ async def execute(
             if self.log_handler:
                 log_interceptor.teardown()
 
-    @override
     async def stream(
         self,
         input: Optional[dict[str, Any]] = None,
@@ -209,6 +212,6 @@ async def stream(
             if self.log_handler:
                 log_interceptor.teardown()
 
-    def cleanup(self) -> None:
-        """Close runtime resources."""
-        pass
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        """Passthrough schema for the delegate."""
+        return await self.delegate.get_schema()

src/uipath/runtime/debug/__init__.py

@@ -1,13 +1,13 @@
 """Initialization module for the debug package."""
 
-from uipath.runtime.debug.bridge import UiPathDebugBridge
+from uipath.runtime.debug.bridge import UiPathDebugBridgeProtocol
 from uipath.runtime.debug.exception import (
     UiPathDebugQuitError,
 )
 from uipath.runtime.debug.runtime import UiPathDebugRuntime
 
 __all__ = [
     "UiPathDebugQuitError",
-    "UiPathDebugBridge",
+    "UiPathDebugBridgeProtocol",
     "UiPathDebugRuntime",
 ]

src/uipath/runtime/debug/bridge.py

@@ -1,7 +1,6 @@
 """Abstract debug bridge interface."""
 
-from abc import ABC, abstractmethod
-from typing import Any, Literal
+from typing import Any, Literal, Protocol
 
 from uipath.runtime import (
     UiPathBreakpointResult,
@@ -10,65 +9,56 @@
 from uipath.runtime.events import UiPathRuntimeStateEvent
 
 
-class UiPathDebugBridge(ABC):
+class UiPathDebugBridgeProtocol(Protocol):
     """Abstract interface for debug communication.
 
     Implementations: SignalR, Console, WebSocket, etc.
     """
 
-    @abstractmethod
     async def connect(self) -> None:
         """Establish connection to debugger."""
-        pass
+        ...
 
-    @abstractmethod
     async def disconnect(self) -> None:
         """Close connection to debugger."""
-        pass
+        ...
 
-    @abstractmethod
     async def emit_execution_started(self, **kwargs) -> None:
         """Notify debugger that execution started."""
-        pass
+        ...
 
-    @abstractmethod
     async def emit_state_update(self, state_event: UiPathRuntimeStateEvent) -> None:
         """Notify debugger of runtime state update."""
-        pass
+        ...
 
-    @abstractmethod
     async def emit_breakpoint_hit(
         self, breakpoint_result: UiPathBreakpointResult
     ) -> None:
         """Notify debugger that a breakpoint was hit."""
-        pass
+        ...
 
-    @abstractmethod
     async def emit_execution_completed(
         self,
         runtime_result: UiPathRuntimeResult,
     ) -> None:
         """Notify debugger that execution completed."""
-        pass
+        ...
 
-    @abstractmethod
     async def emit_execution_error(
         self,
         error: str,
     ) -> None:
         """Notify debugger that an error occurred."""
-        pass
+        ...
 
-    @abstractmethod
     async def wait_for_resume(self) -> Any:
         """Wait for resume command from debugger."""
-        pass
+        ...
 
-    @abstractmethod
     def get_breakpoints(self) -> list[str] | Literal["*"]:
         """Get nodes to suspend execution at.
 
         Returns:
             List of node names to suspend at, or ["*"] for all nodes (step mode)
         """
-        pass
+        ...

src/uipath/runtime/debug/runtime.py

@@ -1,39 +1,38 @@
 """Debug runtime implementation."""
 
 import logging
-from typing import Any, Generic, Optional, TypeVar
+from typing import Any, Optional
 
 from uipath.runtime import (
-    UiPathBaseRuntime,
     UiPathBreakpointResult,
     UiPathExecuteOptions,
+    UiPathRuntimeProtocol,
     UiPathRuntimeResult,
     UiPathRuntimeStatus,
     UiPathStreamNotSupportedError,
     UiPathStreamOptions,
 )
-from uipath.runtime.debug import UiPathDebugBridge, UiPathDebugQuitError
+from uipath.runtime.debug import UiPathDebugBridgeProtocol, UiPathDebugQuitError
 from uipath.runtime.events import (
     UiPathRuntimeStateEvent,
 )
+from uipath.runtime.schema import UiPathRuntimeSchema
 
 logger = logging.getLogger(__name__)
 
-T = TypeVar("T", bound=UiPathBaseRuntime)
 
-
-class UiPathDebugRuntime(UiPathBaseRuntime, Generic[T]):
+class UiPathDebugRuntime:
     """Specialized runtime for debug runs that streams events to a debug bridge."""
 
     def __init__(
         self,
-        delegate: T,
-        debug_bridge: UiPathDebugBridge,
+        delegate: UiPathRuntimeProtocol,
+        debug_bridge: UiPathDebugBridgeProtocol,
     ):
         """Initialize the UiPathDebugRuntime."""
         super().__init__()
-        self.delegate: T = delegate
-        self.debug_bridge: UiPathDebugBridge = debug_bridge
+        self.delegate = delegate
+        self.debug_bridge: UiPathDebugBridgeProtocol = debug_bridge
 
     async def execute(
         self,
@@ -129,3 +128,7 @@ async def cleanup(self) -> None:
             await self.debug_bridge.disconnect()
         except Exception as e:
             logger.warning(f"Error disconnecting debug bridge: {e}")
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        """Passthrough schema for the delegate."""
+        return await self.delegate.get_schema()

src/uipath/runtime/factory.py

@@ -1,25 +1,21 @@
-"""Factory for creating UiPath runtime instances."""
+"""Protocols for creating UiPath runtime instances."""
 
-from abc import ABC, abstractmethod
-from typing import (
-    Generic,
-    TypeVar,
-)
+from typing import Protocol
 
-from uipath.runtime.base import UiPathBaseRuntime
+from uipath.runtime.base import UiPathRuntimeProtocol
 
-T = TypeVar("T", bound=UiPathBaseRuntime)
 
+class UiPathRuntimeScanner(Protocol):
+    """Protocol for discovering all UiPath runtime instances."""
 
-class UiPathRuntimeFactory(Generic[T], ABC):
-    """Generic factory for UiPath runtime classes."""
-
-    @abstractmethod
-    def discover_runtimes(self) -> list[T]:
+    def discover_runtimes(self) -> list[UiPathRuntimeProtocol]:
         """Discover all runtime classes."""
-        raise NotImplementedError()
+        ...
+
+
+class UiPathRuntimeCreator(Protocol):
+    """Protocol for creating a UiPath runtime given an entrypoint."""
 
-    @abstractmethod
-    def new_runtime(self, entrypoint: str) -> T:
+    def new_runtime(self, entrypoint: str) -> UiPathRuntimeProtocol:
         """Create a new runtime instance."""
-        raise NotImplementedError()
+        ...