fix: Rewrite Type Handling

Move type handling logic from activity.py to fn_signature.py, providing utilities for understanding a functions parameters and converting calls/payloads to match the function's signature.

Use this new logic in Workflows, to support any number of typed parameters.

Rewrite the Activity decorators to be fully type-safe. Introduce `@activity.method` and `@activity.impl` to satisfy type safety while allowing for interface types or classes that contain activities.

Fix type-safety for Workflows. This only addresses the annotations and doesn't provide a type-safe invocation mechanism like what is supported with Activities.

Author: natemort

cadence/_internal/activity/__init__.py

@@ -1,5 +1,10 @@
 from ._activity_executor import ActivityExecutor
+from ._definition import AsyncImpl, AsyncMethodImpl, SyncImpl, SyncMethodImpl
 
 __all__ = [
     "ActivityExecutor",
+    "AsyncImpl",
+    "AsyncMethodImpl",
+    "SyncImpl",
+    "SyncMethodImpl",
 ]

cadence/_internal/activity/_activity_executor.py

@@ -1,12 +1,13 @@
 from concurrent.futures import ThreadPoolExecutor
 from logging import getLogger
 from traceback import format_exception
-from typing import Any, Callable
+from typing import Any, Callable, cast
 from google.protobuf.duration import to_timedelta
 from google.protobuf.timestamp import to_datetime
 
 from cadence._internal.activity._context import _Context, _SyncContext
-from cadence.activity import ActivityInfo, ActivityDefinition, ExecutionStrategy
+from cadence._internal.activity._definition import BaseDefinition, ExecutionStrategy
+from cadence.activity import ActivityInfo, ActivityDefinition
 from cadence.api.v1.common_pb2 import Failure
 from cadence.api.v1.service_worker_pb2 import (
     PollForActivityTaskResponse,
@@ -42,12 +43,13 @@ async def execute(self, task: PollForActivityTaskResponse):
             result = await context.execute(task.input)
             await self._report_success(task, result)
         except Exception as e:
+            _logger.exception("Activity failed")
             await self._report_failure(task, e)
 
     def _create_context(self, task: PollForActivityTaskResponse) -> _Context:
         activity_type = task.activity_type.name
         try:
-            activity_def = self._registry(activity_type)
+            activity_def = cast(BaseDefinition, self._registry(activity_type))
         except KeyError:
             raise KeyError(f"Activity type not found: {activity_type}") from None
 

cadence/_internal/activity/_context.py

@@ -3,7 +3,8 @@
 from typing import Any
 
 from cadence import Client
-from cadence.activity import ActivityInfo, ActivityContext, ActivityDefinition
+from cadence._internal.activity._definition import BaseDefinition
+from cadence.activity import ActivityInfo, ActivityContext
 from cadence.api.v1.common_pb2 import Payload
 
 
@@ -12,20 +13,21 @@ def __init__(
         self,
         client: Client,
         info: ActivityInfo,
-        activity_fn: ActivityDefinition[[Any], Any],
+        activity_def: BaseDefinition[[Any], Any],
     ):
         self._client = client
         self._info = info
-        self._activity_fn = activity_fn
+        self._activity_def = activity_def
 
     async def execute(self, payload: Payload) -> Any:
         params = self._to_params(payload)
         with self._activate():
-            return await self._activity_fn(*params)
+            return await self._activity_def.impl_fn(*params)
 
     def _to_params(self, payload: Payload) -> list[Any]:
-        type_hints = [param.type_hint for param in self._activity_fn.params]
-        return self._client.data_converter.from_data(payload, type_hints)
+        return self._activity_def.signature.params_from_payload(
+            self._client.data_converter, payload
+        )
 
     def client(self) -> Client:
         return self._client
@@ -39,10 +41,10 @@ def __init__(
         self,
         client: Client,
         info: ActivityInfo,
-        activity_fn: ActivityDefinition[[Any], Any],
+        activity_def: BaseDefinition[[Any], Any],
         executor: ThreadPoolExecutor,
     ):
-        super().__init__(client, info, activity_fn)
+        super().__init__(client, info, activity_def)
         self._executor = executor
 
     async def execute(self, payload: Payload) -> Any:
@@ -52,7 +54,7 @@ async def execute(self, payload: Payload) -> Any:
 
     def _run(self, args: list[Any]) -> Any:
         with self._activate():
-            return self._activity_fn(*args)
+            return self._activity_def.impl_fn(*args)
 
     def client(self) -> Client:
         raise RuntimeError("client is only supported in async activities")

cadence/_internal/activity/_definition.py

@@ -0,0 +1,178 @@
+import abc
+from abc import ABC
+from enum import Enum
+from functools import update_wrapper, partial
+from typing import (
+    Generic,
+    Callable,
+    Unpack,
+    Self,
+    ParamSpec,
+    TypeVar,
+    Awaitable,
+    cast,
+    Concatenate,
+)
+
+from cadence._internal.fn_signature import FnSignature
+from cadence.workflow import ActivityOptions, WorkflowContext, execute_activity
+
+T = TypeVar("T")
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+class ExecutionStrategy(Enum):
+    ASYNC = "async"
+    THREAD_POOL = "thread_pool"
+
+
+class BaseDefinition(ABC, Generic[P, R]):
+    def __init__(
+        self,
+        name: str,
+        wrapped: Callable,
+        strategy: ExecutionStrategy,
+        signature: FnSignature,
+    ):
+        self._name = name
+        self._wrapped = wrapped
+        self._strategy = strategy
+        self._signature = signature
+        self._execution_options = ActivityOptions()
+
+    @property
+    def strategy(self) -> ExecutionStrategy:
+        return self._strategy
+
+    @property
+    def signature(self) -> FnSignature:
+        return self._signature
+
+    @property
+    def impl_fn(self) -> Callable:
+        return self._wrapped
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @abc.abstractmethod
+    def clone(self) -> Self: ...
+
+    def rebind(self, fn: Callable) -> Self:
+        res = self.clone()
+        res._wrapped = fn
+        return res
+
+    def with_options(self, **kwargs: Unpack[ActivityOptions]) -> Self:
+        res = self.clone()
+        new_opts = self._execution_options.copy()
+        new_opts.update(kwargs)
+        res._execution_options = new_opts
+        return res
+
+    async def execute(self, *args: P.args, **kwargs: P.kwargs) -> R:
+        result_type = cast(type[R], self._signature.return_type)
+        return await execute_activity(
+            self._name,
+            result_type,
+            *self._signature.params_from_call(args, kwargs),
+            **self._execution_options,
+        )
+
+
+class SyncImpl(BaseDefinition[P, R]):
+    def __init__(
+        self,
+        wrapped: Callable[P, R],
+        name: str,
+        signature: FnSignature,
+    ):
+        super().__init__(name, wrapped, ExecutionStrategy.THREAD_POOL, signature)
+        update_wrapper(self, wrapped)
+
+    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
+        if WorkflowContext.is_set():
+            raise RuntimeError(
+                "Attempting to invoke sync function in workflow. Use execute"
+            )
+        return self._wrapped(*args, **kwargs)  # type: ignore
+
+    def clone(self) -> "SyncImpl[P, R]":
+        return SyncImpl[P, R](self._wrapped, self._name, self._signature)
+
+
+class SyncMethodImpl(BaseDefinition[P, R], Generic[T, P, R]):
+    def __init__(
+        self,
+        wrapped: Callable[Concatenate[T, P], R],
+        name: str,
+        signature: FnSignature,
+    ):
+        super().__init__(name, wrapped, ExecutionStrategy.THREAD_POOL, signature)
+        update_wrapper(self, wrapped)
+
+    def __get__(self, instance, owner):
+        if instance is None:
+            return self
+        # If we bound the method to an instance, then drop the self parameter. It's a normal function again
+        return SyncImpl[P, R](
+            partial(self._wrapped, instance), self.name, self._signature
+        )
+
+    def __call__(self, original_self: T, *args: P.args, **kwargs: P.kwargs) -> R:
+        if WorkflowContext.is_set():
+            raise RuntimeError(
+                "Attempting to invoke sync function in workflow. Use execute"
+            )
+        return self._wrapped(original_self, *args, **kwargs)  # type: ignore
+
+    def clone(self) -> "SyncMethodImpl[T, P, R]":
+        return SyncMethodImpl[T, P, R](self._wrapped, self._name, self._signature)
+
+
+class AsyncImpl(BaseDefinition[P, R]):
+    def __init__(
+        self,
+        wrapped: Callable[P, Awaitable[R]],
+        name: str,
+        signature: FnSignature,
+    ):
+        super().__init__(name, wrapped, ExecutionStrategy.ASYNC, signature)
+        update_wrapper(self, wrapped)
+
+    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
+        if WorkflowContext.is_set():
+            return await self.execute(*args, **kwargs)  # type: ignore
+        return await self._wrapped(*args, **kwargs)  # type: ignore
+
+    def clone(self) -> "AsyncImpl[P, R]":
+        return AsyncImpl[P, R](self._wrapped, self._name, self._signature)
+
+
+class AsyncMethodImpl(BaseDefinition[P, R], Generic[T, P, R]):
+    def __init__(
+        self,
+        wrapped: Callable[Concatenate[T, P], Awaitable[R]],
+        name: str,
+        signature: FnSignature,
+    ):
+        super().__init__(name, wrapped, ExecutionStrategy.ASYNC, signature)
+        update_wrapper(self, wrapped)
+
+    def __get__(self, instance, owner):
+        if instance is None:
+            return self
+        # If we bound the method to an instance, then drop the self parameter. It's a normal function again
+        return AsyncImpl[P, R](
+            partial(self._wrapped, instance), self.name, self._signature
+        )
+
+    async def __call__(self, original_self: T, *args: P.args, **kwargs: P.kwargs) -> R:
+        if WorkflowContext.is_set():
+            return await self.execute(*args, **kwargs)  # type: ignore
+        return await self._wrapped(original_self, *args, **kwargs)  # type: ignore
+
+    def clone(self) -> "AsyncMethodImpl[T, P, R]":
+        return AsyncMethodImpl[T, P, R](self._wrapped, self._name, self._signature)

cadence/_internal/fn_signature.py

@@ -0,0 +1,90 @@
+from dataclasses import dataclass
+from inspect import signature, Parameter
+from typing import (
+    Type,
+    Any,
+    Callable,
+    Sequence,
+    get_type_hints,
+)
+
+from cadence.api.v1.common_pb2 import Payload
+from cadence.data_converter import DataConverter
+
+
+@dataclass(frozen=True)
+class FnParameter:
+    name: str
+    type_hint: Type | None
+    has_default: bool = False
+    default_value: Any = None
+
+
+@dataclass(frozen=True)
+class FnSignature:
+    params: list[FnParameter]
+    return_type: Type
+
+    def params_from_call(
+        self, args: Sequence[Any], kwargs: dict[str, Any]
+    ) -> list[Any]:
+        result: list[Any] = []
+        if len(args) > len(self.params):
+            raise ValueError(f"Too many positional arguments: {args}")
+
+        for value, param_spec in zip(args, self.params):
+            result.append(value)
+
+        i = len(result)
+        while i < len(self.params):
+            param = self.params[i]
+            if param.name not in kwargs and not param.has_default:
+                raise ValueError(f"Missing parameter: {param.name}")
+
+            value = kwargs.pop(param.name, param.default_value)
+            result.append(value)
+            i = i + 1
+
+        if len(kwargs) > 0:
+            raise ValueError(f"Unexpected keyword arguments: {kwargs}")
+
+        return result
+
+    def params_from_payload(
+        self, data_converter: DataConverter, payload: Payload
+    ) -> list[Any]:
+        type_hints = [param.type_hint for param in self.params]
+        return data_converter.from_data(payload, type_hints)
+
+    @staticmethod
+    def of(fn: Callable) -> "FnSignature":
+        sig = signature(fn)
+        args = sig.parameters
+        hints = get_type_hints(fn)
+        params = []
+        for name, param in args.items():
+            # "unbound functions" aren't a thing in the Python spec. We don't have a way to determine whether the function
+            # is part of a class or is standalone.
+            # Filter out the self parameter and hope they followed the convention.
+            if param.name == "self":
+                continue
+            default = None
+            has_default = False
+            if param.default != Parameter.empty:
+                default = param.default
+                has_default = param.default is not None
+            if param.kind in (
+                Parameter.POSITIONAL_ONLY,
+                Parameter.POSITIONAL_OR_KEYWORD,
+            ):
+                type_hint = hints.get(name, None)
+                params.append(FnParameter(name, type_hint, has_default, default))
+            else:
+                raise ValueError(
+                    f"Parameters must be positional. {name} is {param.kind}, and not valid"
+                )
+
+        # Treat unspecified return type as Any
+        return_type = hints.get("return", Any)
+
+        return FnSignature(params, return_type)

cadence/_internal/workflow/workflow_intance.py

@@ -1,5 +1,5 @@
 from asyncio import CancelledError, InvalidStateError, Task
-from typing import Any, Optional
+from typing import Optional
 from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop
 from cadence.api.v1.common_pb2 import Payload
 from cadence.data_converter import DataConverter
@@ -20,11 +20,13 @@ def __init__(
         self._instance = workflow_definition.cls()  # construct a new workflow object
         self._task: Optional[Task] = None
 
-    def start(self, input: Payload):
+    def start(self, payload: Payload):
         if self._task is None:
             run_method = self._definition.get_run_method(self._instance)
-            # TODO handle multiple inputs
-            workflow_input = self._data_converter.from_data(input, [Any])
+            # noinspection PyProtectedMember
+            workflow_input = self._definition._run_signature.params_from_payload(
+                self._data_converter, payload
+            )
             self._task = self._loop.create_task(run_method(*workflow_input))
 
     def run_once(self):