Outcome.intercept NoResult (#174)

The original implementation of `Outcome.intercept` used `Optional[T]`
and thus could not distingish between no cached results and a cached
result of None.

This PR adds a singleton NoResult class to use instead of `None` in as
the return value of the `interceptor` parameter of `Outcome.intercept`.

Author: devhawk

dbos/_core.py

@@ -21,7 +21,7 @@
     overload,
 )
 
-from dbos._outcome import Immediate, Outcome, Pending
+from dbos._outcome import Immediate, NoResult, Outcome, Pending
 
 from ._app_db import ApplicationDatabase, TransactionResultInternal
 
@@ -719,7 +719,7 @@ def record_step_result(func: Callable[[], R]) -> R:
                 finally:
                     dbos._sys_db.record_operation_result(step_output)
 
-            def check_existing_result() -> Optional[str]:
+            def check_existing_result() -> Union[NoResult, R]:
                 ctx = assert_current_dbos_context()
                 recorded_output = dbos._sys_db.check_operation_execution(
                     ctx.workflow_id, ctx.function_id
@@ -734,14 +734,16 @@ def check_existing_result() -> Optional[str]:
                         )
                         raise deserialized_error
                     elif recorded_output["output"] is not None:
-                        return recorded_output["output"]
+                        return cast(
+                            R, _serialization.deserialize(recorded_output["output"])
+                        )
                     else:
                         raise Exception("Output and error are both None")
                 else:
                     dbos.logger.debug(
                         f"Running step, id: {ctx.function_id}, name: {attributes['name']}"
                     )
-                    return None
+                    return NoResult()
 
             stepOutcome = Outcome[R].make(functools.partial(func, *args, **kwargs))
             if retries_allowed:

dbos/_outcome.py

@@ -4,12 +4,20 @@
 import time
 from typing import Any, Callable, Coroutine, Optional, Protocol, TypeVar, Union, cast
 
-from . import _serialization
-
 T = TypeVar("T")
 R = TypeVar("R")
 
 
+class NoResult:
+    _instance: Optional["NoResult"] = None
+    __slots__ = ()
+
+    def __new__(cls, *args: Any, **kwargs: Any) -> "NoResult":
+        if not cls._instance:
+            cls._instance = super(NoResult, cls).__new__(cls, *args, **kwargs)
+        return cls._instance
+
+
 # define Outcome protocol w/ common composition methods
 class Outcome(Protocol[T]):
 
@@ -30,7 +38,9 @@ def retry(
         exceeded_retries: Callable[[int], BaseException],
     ) -> "Outcome[T]": ...
 
-    def intercept(self, interceptor: Callable[[], Optional[str]]) -> "Outcome[T]": ...
+    def intercept(
+        self, interceptor: Callable[[], Union[NoResult, T]]
+    ) -> "Outcome[T]": ...
 
     def __call__(self) -> Union[T, Coroutine[Any, Any, T]]: ...
 
@@ -61,14 +71,14 @@ def wrap(
 
     @staticmethod
     def _intercept(
-        func: Callable[[], T], interceptor: Callable[[], Optional[str]]
+        func: Callable[[], T], interceptor: Callable[[], Union[NoResult, T]]
     ) -> T:
         intercepted = interceptor()
-        return (
-            cast(T, _serialization.deserialize(intercepted)) if intercepted else func()
-        )
+        return intercepted if not isinstance(intercepted, NoResult) else func()
 
-    def intercept(self, interceptor: Callable[[], Optional[str]]) -> "Immediate[T]":
+    def intercept(
+        self, interceptor: Callable[[], Union[NoResult, T]]
+    ) -> "Immediate[T]":
         return Immediate[T](lambda: Immediate._intercept(self._func, interceptor))
 
     @staticmethod
@@ -157,16 +167,12 @@ def also(self, cm: contextlib.AbstractContextManager[Any, bool]) -> "Pending[T]"
     @staticmethod
     async def _intercept(
         func: Callable[[], Coroutine[Any, Any, T]],
-        interceptor: Callable[[], Optional[str]],
+        interceptor: Callable[[], Union[NoResult, T]],
     ) -> T:
         intercepted = await asyncio.to_thread(interceptor)
-        return (
-            cast(T, _serialization.deserialize(intercepted))
-            if intercepted
-            else await func()
-        )
+        return intercepted if not isinstance(intercepted, NoResult) else await func()
 
-    def intercept(self, interceptor: Callable[[], Optional[str]]) -> "Pending[T]":
+    def intercept(self, interceptor: Callable[[], Union[NoResult, T]]) -> "Pending[T]":
         return Pending[T](lambda: Pending._intercept(self._func, interceptor))
 
     @staticmethod