More tests

Author: cretz

README.md

@@ -467,10 +467,8 @@ synchronous multithreaded activities.
 
 ###### Synchronous Multiprocess/Other Activities
 
-Synchronous activities, i.e. functions that do not have `async def`, can be used with workers, but the
-`activity_executor` worker parameter must be set with a `concurrent.futures.Executor` instance to use for executing the
-activities. If this is _not_ set to an instance of `concurrent.futures.ThreadPoolExecutor` then the synchronous
-activities are considered multiprocess/other activities.
+If `activity_executor` is set to an instance of `concurrent.futures.Executor` that is _not_
+`concurrent.futures.ThreadPoolExecutor`, then the synchronous activities are considered multiprocess/other activities.
 
 These require special primitives for heartbeating and cancellation. The `shared_state_manager` worker parameter must be
 set to an instance of `temporalio.worker.SharedStateManager`. The most common implementation can be created by passing a

pyproject.toml

@@ -90,7 +90,7 @@ asyncio_mode = "auto"
 log_cli = true
 log_cli_level = "INFO"
 log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)"
-timeout = 300
+timeout = 600
 timeout_func_only = true
 
 [tool.isort]

temporalio/client.py

@@ -347,14 +347,13 @@ async def start_workflow(
         """
         # Use definition if callable
         name: str
-        arg_types: Optional[List[Type]] = None
         ret_type: Optional[Type] = None
         if isinstance(workflow, str):
             name = workflow
         elif callable(workflow):
             defn = temporalio.workflow._Definition.must_from_run_fn(workflow)
             name = defn.name
-            arg_types, ret_type = self._type_lookup.get_type_hints(defn.run_fn)
+            _, ret_type = self._type_lookup.get_type_hints(defn.run_fn)
         else:
             raise TypeError("Workflow must be a string or callable")
 
@@ -375,7 +374,6 @@ async def start_workflow(
                 header=header,
                 start_signal=start_signal,
                 start_signal_args=start_signal_args,
-                arg_types=arg_types,
                 ret_type=ret_type,
             )
         )
@@ -646,13 +644,15 @@ def __init__(
         run_id: Optional[str] = None,
         result_run_id: Optional[str] = None,
         first_execution_run_id: Optional[str] = None,
+        result_type: Optional[Type] = None,
     ) -> None:
         """Create workflow handle."""
         self._client = client
         self._id = id
         self._run_id = run_id
         self._result_run_id = result_run_id
         self._first_execution_run_id = first_execution_run_id
+        self._result_type = result_type
 
     @property
     def id(self) -> str:
@@ -754,9 +754,10 @@ async def result(self, *, follow_runs: bool = True) -> WorkflowReturnType:
                     req.next_page_token = b""
                     continue
                 # Ignoring anything after the first response like TypeScript
-                # TODO(cretz): Support type hints
+                type_hints = [self._result_type] if self._result_type else None
                 results = await self._client.data_converter.decode_wrapper(
-                    complete_attr.result
+                    complete_attr.result,
+                    type_hints,
                 )
                 if not results:
                     return cast(WorkflowReturnType, None)
@@ -969,6 +970,7 @@ async def query(
             RPCError: Workflow details could not be fetched.
         """
         query_name: str
+        ret_type: Optional[Type] = None
         if callable(query):
             defn = temporalio.workflow._QueryDefinition.from_fn(query)
             if not defn:
@@ -980,6 +982,7 @@ async def query(
                 raise RuntimeError("Cannot invoke dynamic query definition")
             # TODO(cretz): Check count/type of args at runtime?
             query_name = defn.name
+            _, ret_type = self._client._type_lookup.get_type_hints(defn.fn)
         else:
             query_name = str(query)
 
@@ -991,6 +994,7 @@ async def query(
                 args=temporalio.common._arg_or_args(arg, args),
                 reject_condition=reject_condition
                 or self._client._config["default_workflow_query_reject_condition"],
+                ret_type=ret_type,
             )
         )
 
@@ -1245,8 +1249,7 @@ class StartWorkflowInput:
     header: Optional[Mapping[str, Any]]
     start_signal: Optional[str]
     start_signal_args: Iterable[Any]
-    # The types may be absent
-    arg_types: Optional[List[Type]]
+    # Type may be absent
     ret_type: Optional[Type]
 
 
@@ -1268,6 +1271,8 @@ class QueryWorkflowInput:
     query: str
     args: Iterable[Any]
     reject_condition: Optional[temporalio.common.QueryRejectCondition]
+    # Type may be absent
+    ret_type: Optional[Type]
 
 
 @dataclass
@@ -1429,6 +1434,7 @@ async def start_workflow(
             req.workflow_id,
             result_run_id=resp.run_id,
             first_execution_run_id=first_execution_run_id,
+            result_type=input.ret_type,
         )
 
     async def cancel_workflow(self, input: CancelWorkflowInput) -> None:
@@ -1474,7 +1480,10 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any:
             )
         if not resp.query_result.payloads:
             return None
-        results = await self._client.data_converter.decode(resp.query_result.payloads)
+        type_hints = [input.ret_type] if input.ret_type else None
+        results = await self._client.data_converter.decode(
+            resp.query_result.payloads, type_hints
+        )
         if not results:
             return None
         elif len(results) > 1:

temporalio/converter.py

@@ -571,14 +571,16 @@ async def encode_wrapper(
         return temporalio.api.common.v1.Payloads(payloads=(await self.encode(values)))
 
     async def decode_wrapper(
-        self, payloads: Optional[temporalio.api.common.v1.Payloads]
+        self,
+        payloads: Optional[temporalio.api.common.v1.Payloads],
+        type_hints: Optional[List[Type]] = None,
     ) -> List[Any]:
         """:py:meth:`decode` for the
         :py:class:`temporalio.api.common.v1.Payloads` wrapper.
         """
         if not payloads or not payloads.payloads:
             return []
-        return await self.decode(payloads.payloads)
+        return await self.decode(payloads.payloads, type_hints)
 
 
 _default: Optional[DataConverter] = None

temporalio/worker/activity.py

@@ -356,7 +356,7 @@ async def _run_activity(
                 heartbeat_timeout=start.heartbeat_timeout.ToTimedelta()
                 if start.HasField("heartbeat_timeout")
                 else None,
-                is_local=False,
+                is_local=start.is_local,
                 retry_policy=temporalio.bridge.worker.retry_policy_from_proto(
                     start.retry_policy
                 )

temporalio/worker/worker.py

@@ -224,7 +224,6 @@ def __init__(
                 data_converter=client_config["data_converter"],
                 interceptors=interceptors,
                 type_hint_eval_str=client_config["type_hint_eval_str"],
-                max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
             )
 
         # Create bridge worker last. We have empirically observed that if it is

temporalio/worker/workflow.py

@@ -53,7 +53,6 @@ def __init__(
         data_converter: temporalio.converter.DataConverter,
         interceptors: Iterable[Interceptor],
         type_hint_eval_str: bool,
-        max_concurrent_workflow_tasks: int,
     ) -> None:
         self._bridge_worker = bridge_worker
         self._namespace = namespace
@@ -182,6 +181,9 @@ async def _handle_activation(
                     f"Failed converting activation exception: {inner_err}"
                 )
 
+        # Always set the run ID on the completion
+        completion.run_id = act.run_id
+
         # Encode the completion if there's a codec
         if self._data_converter.payload_codec:
             try:
@@ -234,6 +236,13 @@ async def _create_workflow_instance(
 
         # Build info
         start = start_job.start_workflow
+        parent: Optional[temporalio.workflow.ParentInfo] = None
+        if start.HasField("parent_workflow_info"):
+            parent = temporalio.workflow.ParentInfo(
+                namespace=start.parent_workflow_info.namespace,
+                run_id=start.parent_workflow_info.run_id,
+                workflow_id=start.parent_workflow_info.workflow_id,
+            )
         info = temporalio.workflow.Info(
             attempt=start.attempt,
             continued_run_id=start.continued_from_execution_run_id or None,
@@ -242,6 +251,7 @@ async def _create_workflow_instance(
             if start.HasField("workflow_execution_timeout")
             else None,
             namespace=self._namespace,
+            parent=parent,
             run_id=act.run_id,
             run_timeout=start.workflow_run_timeout.ToTimedelta()
             if start.HasField("workflow_run_timeout")

temporalio/worker/workflow_instance.py

@@ -213,8 +213,8 @@ def activate(
         self._current_completion = (
             temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion()
         )
-        self._current_completion.run_id = act.run_id
         self._current_completion.successful.SetInParent()
+        self._current_activation_error: Optional[Exception] = None
         self._time = act.timestamp.ToMicroseconds() / 1e6
         self._is_replaying = act.is_replaying
 
@@ -641,6 +641,15 @@ def workflow_get_query_handler(self, name: Optional[str]) -> Optional[Callable]:
             return cast(Optional[Callable], defn)
         # Curry instance on the definition function since that represents an
         # unbound method
+        if inspect.iscoroutinefunction(defn.fn):
+            # We cannot use functools.partial here because in <= 3.7 that isn't
+            # considered an inspect.iscoroutinefunction
+            fn = cast(Callable[..., Awaitable[Any]], defn.fn)
+
+            async def with_object(*args, **kwargs) -> Any:
+                return await fn(self._object, *args, **kwargs)
+
+            return with_object
         return partial(defn.fn, self._object)
 
     def workflow_get_signal_handler(self, name: Optional[str]) -> Optional[Callable]:
@@ -650,6 +659,15 @@ def workflow_get_signal_handler(self, name: Optional[str]) -> Optional[Callable]
             return cast(Optional[Callable], defn)
         # Curry instance on the definition function since that represents an
         # unbound method
+        if inspect.iscoroutinefunction(defn.fn):
+            # We cannot use functools.partial here because in <= 3.7 that isn't
+            # considered an inspect.iscoroutinefunction
+            fn = cast(Callable[..., Awaitable[Any]], defn.fn)
+
+            async def with_object(*args, **kwargs) -> Any:
+                return await fn(self._object, *args, **kwargs)
+
+            return with_object
         return partial(defn.fn, self._object)
 
     def workflow_info(self) -> temporalio.workflow.Info:
@@ -1008,6 +1026,11 @@ def _run_once(self) -> None:
                     handle = self._ready.popleft()
                     handle._run()
 
+                    # Must throw here. Only really set inside
+                    # _run_top_level_workflow_function.
+                    if self._current_activation_error:
+                        raise self._current_activation_error
+
                 # Check conditions which may add to the ready list
                 self._conditions[:] = [
                     t for t in self._conditions if not self._check_condition(*t)
@@ -1018,8 +1041,6 @@ def _run_once(self) -> None:
     # This is used for the primary workflow function and signal handlers in
     # order to apply common exception handling to each
     async def _run_top_level_workflow_function(self, coro: Awaitable[None]) -> None:
-        # We intentionally don't catch all errors, instead we bubble those
-        # out as task failures
         try:
             await coro
         except _ContinueAsNewError as err:
@@ -1048,6 +1069,8 @@ async def _run_top_level_workflow_function(self, coro: Awaitable[None]) -> None:
                 self._payload_converter,
                 command.fail_workflow_execution.failure,
             )
+        except Exception as err:
+            self._current_activation_error = err
 
     async def _signal_external_workflow(
         self,
@@ -1514,10 +1537,14 @@ def _resolve_success(self, result: Any) -> None:
         self._result_fut.set_result(result)
 
     def _resolve_failure(self, err: Exception) -> None:
-        if not self._start_fut.done():
+        if self._start_fut.done():
+            # We intentionally let this error if already done
+            self._result_fut.set_exception(err)
+        else:
             self._start_fut.set_exception(err)
-        # We intentionally let this error if already done
-        self._result_fut.set_exception(err)
+            # Set the result as none to avoid Python warning about unhandled
+            # future
+            self._result_fut.set_result(None)
 
     def _apply_start_command(
         self,

temporalio/workflow.py

@@ -265,6 +265,7 @@ class Info:
     cron_schedule: Optional[str]
     execution_timeout: Optional[timedelta]
     namespace: str
+    parent: Optional[ParentInfo]
     run_id: str
     run_timeout: Optional[timedelta]
     start_time: datetime
@@ -274,9 +275,6 @@ class Info:
     workflow_type: str
 
     # TODO(cretz): memo
-    # TODO(cretz): parent_namespace
-    # TODO(cretz): parent_run_id
-    # TODO(cretz): parent_workflow_id
     # TODO(cretz): retry_policy
     # TODO(cretz): search_attributes
 
@@ -288,6 +286,15 @@ def _logger_details(self) -> Mapping[str, Any]:
         }
 
 
+@dataclass(frozen=True)
+class ParentInfo:
+    """Information about the parent workflow."""
+
+    namespace: str
+    run_id: str
+    workflow_id: str
+
+
 class _Runtime(ABC):
     @staticmethod
     def current() -> _Runtime:
@@ -548,7 +555,11 @@ class _Definition:
 
     @staticmethod
     def from_class(cls: Type) -> Optional[_Definition]:
-        return getattr(cls, "__temporal_workflow_definition", None)
+        # We make sure to only return it if it's on _this_ class
+        defn = getattr(cls, "__temporal_workflow_definition", None)
+        if defn and defn.cls == cls:
+            return defn
+        return None
 
     @staticmethod
     def must_from_class(cls: Type) -> _Definition:
@@ -576,7 +587,8 @@ def must_from_run_fn(fn: Callable[..., Awaitable[Any]]) -> _Definition:
 
     @staticmethod
     def _apply_to_class(cls: Type, workflow_name: str) -> None:
-        if hasattr(cls, "__temporal_workflow_definition"):
+        # Check it's not being doubly applied
+        if _Definition.from_class(cls):
             raise ValueError("Class already contains workflow definition")
         issues: List[str] = []