Support scheduling Activities within Workflows

Author: natemort

cadence/_internal/__init__.py

@@ -6,5 +6,3 @@
 """
 
 __all__: list[str] = []
-
-

cadence/_internal/activity/_activity_executor.py

@@ -8,21 +8,33 @@
 from cadence._internal.activity._context import _Context, _SyncContext
 from cadence.activity import ActivityInfo, ActivityDefinition, ExecutionStrategy
 from cadence.api.v1.common_pb2 import Failure
-from cadence.api.v1.service_worker_pb2 import PollForActivityTaskResponse, RespondActivityTaskFailedRequest, \
-    RespondActivityTaskCompletedRequest
+from cadence.api.v1.service_worker_pb2 import (
+    PollForActivityTaskResponse,
+    RespondActivityTaskFailedRequest,
+    RespondActivityTaskCompletedRequest,
+)
 from cadence.client import Client
 
 _logger = getLogger(__name__)
 
+
 class ActivityExecutor:
-    def __init__(self, client: Client, task_list: str, identity: str, max_workers: int, registry: Callable[[str], ActivityDefinition]):
+    def __init__(
+        self,
+        client: Client,
+        task_list: str,
+        identity: str,
+        max_workers: int,
+        registry: Callable[[str], ActivityDefinition],
+    ):
         self._client = client
         self._data_converter = client.data_converter
         self._registry = registry
         self._identity = identity
         self._task_list = task_list
-        self._thread_pool = ThreadPoolExecutor(max_workers=max_workers,
-                                               thread_name_prefix=f'{task_list}-activity-')
+        self._thread_pool = ThreadPoolExecutor(
+            max_workers=max_workers, thread_name_prefix=f"{task_list}-activity-"
+        )
 
     async def execute(self, task: PollForActivityTaskResponse):
         try:
@@ -46,27 +58,33 @@ def _create_context(self, task: PollForActivityTaskResponse) -> _Context:
         else:
             return _SyncContext(self._client, info, activity_def, self._thread_pool)
 
-    async def _report_failure(self, task: PollForActivityTaskResponse, error: Exception):
+    async def _report_failure(
+        self, task: PollForActivityTaskResponse, error: Exception
+    ):
         try:
-            await self._client.worker_stub.RespondActivityTaskFailed(RespondActivityTaskFailedRequest(
-                task_token=task.task_token,
-                failure=_to_failure(error),
-                identity=self._identity,
-            ))
+            await self._client.worker_stub.RespondActivityTaskFailed(
+                RespondActivityTaskFailedRequest(
+                    task_token=task.task_token,
+                    failure=_to_failure(error),
+                    identity=self._identity,
+                )
+            )
         except Exception:
-            _logger.exception('Exception reporting activity failure')
+            _logger.exception("Exception reporting activity failure")
 
     async def _report_success(self, task: PollForActivityTaskResponse, result: Any):
-        as_payload = await self._data_converter.to_data([result])
+        as_payload = self._data_converter.to_data([result])
 
         try:
-            await self._client.worker_stub.RespondActivityTaskCompleted(RespondActivityTaskCompletedRequest(
-                task_token=task.task_token,
-                result=as_payload,
-                identity=self._identity,
-            ))
+            await self._client.worker_stub.RespondActivityTaskCompleted(
+                RespondActivityTaskCompletedRequest(
+                    task_token=task.task_token,
+                    result=as_payload,
+                    identity=self._identity,
+                )
+            )
         except Exception:
-            _logger.exception('Exception reporting activity complete')
+            _logger.exception("Exception reporting activity complete")
 
     def _create_info(self, task: PollForActivityTaskResponse) -> ActivityInfo:
         return ActivityInfo(

cadence/_internal/activity/_context.py

@@ -14,27 +14,28 @@ def __init__(self, client: Client, info: ActivityInfo, activity_fn: ActivityDefi
         self._activity_fn = activity_fn
 
     async def execute(self, payload: Payload) -> Any:
-        params = await self._to_params(payload)
+        params = self._to_params(payload)
         with self._activate():
             return await self._activity_fn(*params)
 
-    async def _to_params(self, payload: Payload) -> list[Any]:
+    def _to_params(self, payload: Payload) -> list[Any]:
         type_hints = [param.type_hint for param in self._activity_fn.params]
-        return await self._client.data_converter.from_data(payload, type_hints)
+        return self._client.data_converter.from_data(payload, type_hints)
 
     def client(self) -> Client:
         return self._client
 
     def info(self) -> ActivityInfo:
         return self._info
 
+
 class _SyncContext(_Context):
     def __init__(self, client: Client, info: ActivityInfo, activity_fn: ActivityDefinition[[Any], Any], executor: ThreadPoolExecutor):
         super().__init__(client, info, activity_fn)
         self._executor = executor
 
     async def execute(self, payload: Payload) -> Any:
-        params = await self._to_params(payload)
+        params = self._to_params(payload)
         loop = asyncio.get_running_loop()
         return await loop.run_in_executor(self._executor, self._run, params)
 
@@ -44,4 +45,3 @@ def _run(self, args: list[Any]) -> Any:
 
     def client(self) -> Client:
         raise RuntimeError("client is only supported in async activities")
-