Review fixes

Author: Sushisource

temporalio/client.py

@@ -26,6 +26,7 @@
     Mapping,
     Optional,
     Sequence,
+    Tuple,
     Type,
     Union,
     cast,
@@ -43,6 +44,7 @@
 import temporalio.api.failure.v1
 import temporalio.api.history.v1
 import temporalio.api.schedule.v1
+import temporalio.api.sdk.v1
 import temporalio.api.taskqueue.v1
 import temporalio.api.update.v1
 import temporalio.api.workflow.v1
@@ -2423,8 +2425,8 @@ async def _from_raw_description(
         description: temporalio.api.workflowservice.v1.DescribeWorkflowExecutionResponse,
         converter: temporalio.converter.DataConverter,
     ) -> WorkflowExecutionDescription:
-        (summ, deets) = await converter._decode_user_metadata(
-            description.execution_config.user_metadata
+        (summ, deets) = await _decode_user_metadata(
+            converter, description.execution_config.user_metadata
         )
         return WorkflowExecutionDescription._from_raw_info(  # type: ignore
             description.workflow_execution_info,
@@ -3547,8 +3549,8 @@ async def _to_proto(
                         for k, v in self.memo.items()
                     },
                 ),
-                user_metadata=await client.data_converter._encode_user_metadata(
-                    self.static_summary, self.static_details
+                user_metadata=await _encode_user_metadata(
+                    client.data_converter, self.static_summary, self.static_details
                 ),
             ),
         )
@@ -5141,8 +5143,8 @@ async def start_workflow(
             temporalio.converter.encode_search_attributes(
                 input.search_attributes, req.search_attributes
             )
-        metadata = await self._client.data_converter._encode_user_metadata(
-            input.static_summary, input.static_details
+        metadata = await _encode_user_metadata(
+            self._client.data_converter, input.static_summary, input.static_details
         )
         if metadata is not None:
             req.user_metadata.CopyFrom(metadata)
@@ -6373,3 +6375,42 @@ def api_key(self, value: Optional[str]) -> None:
         # Update config and perform update
         self.service_client.config.api_key = value
         self.service_client.update_api_key(value)
+
+
+async def _encode_user_metadata(
+    converter: temporalio.converter.DataConverter,
+    summary: Optional[Union[str, temporalio.api.common.v1.Payload]],
+    details: Optional[Union[str, temporalio.api.common.v1.Payload]],
+) -> Optional[temporalio.api.sdk.v1.UserMetadata]:
+    if summary is None and details is None:
+        return None
+    enc_summary = None
+    enc_details = None
+    if summary is not None:
+        if isinstance(summary, str):
+            enc_summary = (await converter.encode([summary]))[0]
+        else:
+            enc_summary = summary
+    if details is not None:
+        if isinstance(details, str):
+            enc_details = (await converter.encode([details]))[0]
+        else:
+            enc_details = details
+    return temporalio.api.sdk.v1.UserMetadata(summary=enc_summary, details=enc_details)
+
+
+async def _decode_user_metadata(
+    converter: temporalio.converter.DataConverter,
+    metadata: Optional[temporalio.api.sdk.v1.UserMetadata],
+) -> Tuple[Optional[str], Optional[str]]:
+    """Returns (summary, details)"""
+    if metadata is None:
+        return None, None
+    return (
+        None
+        if not metadata.HasField("summary")
+        else (await converter.decode([metadata.summary]))[0],
+        None
+        if not metadata.HasField("details")
+        else (await converter.decode([metadata.details]))[0],
+    )

temporalio/converter.py

@@ -1085,29 +1085,6 @@ async def encode_wrapper(
         """
         return temporalio.api.common.v1.Payloads(payloads=(await self.encode(values)))
 
-    async def _encode_user_metadata(
-        self,
-        summary: Optional[Union[str, temporalio.api.common.v1.Payload]],
-        details: Optional[Union[str, temporalio.api.common.v1.Payload]],
-    ) -> Optional[temporalio.api.sdk.v1.UserMetadata]:
-        if summary is None and details is None:
-            return None
-        enc_summary = None
-        enc_details = None
-        if summary is not None:
-            if isinstance(summary, str):
-                enc_summary = (await self.encode([summary]))[0]
-            else:
-                enc_summary = summary
-        if details is not None:
-            if isinstance(details, str):
-                enc_details = (await self.encode([details]))[0]
-            else:
-                enc_details = details
-        return temporalio.api.sdk.v1.UserMetadata(
-            summary=enc_summary, details=enc_details
-        )
-
     async def decode_wrapper(
         self,
         payloads: Optional[temporalio.api.common.v1.Payloads],
@@ -1136,21 +1113,6 @@ async def decode_failure(
             await self.payload_codec.decode_failure(failure)
         return self.failure_converter.from_failure(failure, self.payload_converter)
 
-    async def _decode_user_metadata(
-        self, metadata: Optional[temporalio.api.sdk.v1.UserMetadata]
-    ) -> Tuple[Optional[str], Optional[str]]:
-        """Returns (summary, details)"""
-        if metadata is None:
-            return None, None
-        return (
-            None
-            if not metadata.HasField("summary")
-            else (await self.decode([metadata.summary]))[0],
-            None
-            if not metadata.HasField("details")
-            else (await self.decode([metadata.details]))[0],
-        )
-
 
 DefaultPayloadConverter.default_encoding_payload_converters = (
     BinaryNullPayloadConverter(),

temporalio/worker/_workflow_instance.py

@@ -331,7 +331,7 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
 
         # The current details (as opposed to static details on workflow start), returned in the
         # metadata query
-        self._current_details: str = ""
+        self._current_details = ""
 
     def get_thread_id(self) -> Optional[int]:
         return self._current_thread_id
@@ -1449,9 +1449,13 @@ async def workflow_sleep(
             if summary
             else None
         )
+        fut = self.create_future()
         self._timer_impl(
-            duration, lambda _: None, options=_TimerOptions(user_metadata=user_metadata)
+            duration,
+            _TimerOptions(user_metadata=user_metadata),
+            lambda: fut.set_result(None),
         )
+        await fut
 
     async def workflow_wait_condition(
         self,
@@ -1476,9 +1480,9 @@ async def workflow_wait_condition(
     def workflow_get_current_details(self) -> str:
         return self._current_details
 
-    def workflow_set_current_details(self, description):
+    def workflow_set_current_details(self, details: str):
         self._assert_not_read_only("set current details")
-        self._current_details = description
+        self._current_details = details
 
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".
@@ -2014,24 +2018,24 @@ def _enhanced_stack_trace(self) -> temporalio.api.sdk.v1.EnhancedStackTrace:
     def _temporal_workflow_metadata(self) -> temporalio.api.sdk.v1.WorkflowMetadata:
         query_definitions = [
             temporalio.api.sdk.v1.WorkflowInteractionDefinition(
-                name=qd.name if qd.name is not None else "",
-                description=qd.description if qd.description is not None else "",
+                name=qd.name or "",
+                description=qd.description or "",
             )
             for qd in self._queries.values()
         ]
         query_definitions.sort(key=lambda qd: qd.name)
         signal_definitions = [
             temporalio.api.sdk.v1.WorkflowInteractionDefinition(
-                name=sd.name if sd.name is not None else "",
-                description=sd.description if sd.description is not None else "",
+                name=sd.name or "",
+                description=sd.description or "",
             )
             for sd in self._signals.values()
         ]
         signal_definitions.sort(key=lambda sd: sd.name)
         update_definitions = [
             temporalio.api.sdk.v1.WorkflowInteractionDefinition(
-                name=ud.name if ud.name is not None else "",
-                description=ud.description if ud.description is not None else "",
+                name=ud.name or "",
+                description=ud.description or "",
             )
             for ud in self._updates.values()
         ]
@@ -2050,10 +2054,10 @@ def _temporal_workflow_metadata(self) -> temporalio.api.sdk.v1.WorkflowMetadata:
     def _timer_impl(
         self,
         delay: float,
+        options: _TimerOptions,
         callback: Callable[..., Any],
         *args: Any,
         context: Optional[contextvars.Context] = None,
-        options: Optional[_TimerOptions] = None,
     ):
         self._assert_not_read_only("schedule timer")
         # Delay must be positive
@@ -2062,12 +2066,6 @@ def _timer_impl(
 
         # Create, schedule, and return
         seq = self._next_seq("timer")
-        # If options aren't explicitly passed, attempt to fetch them from the class field,
-        # erasing them afterward. Support callers who cannot call this directly because they
-        # rely on asyncio functions.
-        if options is None:
-            options = self._next_timer_options
-            self._next_timer_options = None
         handle = _TimerHandle(
             seq, self.time() + delay, options, callback, args, self, context
         )
@@ -2079,7 +2077,6 @@ def _timer_impl(
     # These are in the order defined in CPython's impl of the base class. Many
     # functions are intentionally not implemented/supported.
 
-    # TODO/Review: This doesn't appear to implement any base class fn and isn't called anywhere?
     def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None:
         if not isinstance(handle, _TimerHandle):
             raise TypeError("Expected Temporal timer handle")
@@ -2107,7 +2104,12 @@ def call_later(
         *args: Any,
         context: Optional[contextvars.Context] = None,
     ) -> asyncio.TimerHandle:
-        return self._timer_impl(delay, callback, *args, context=context)
+        # Fetch options from the class field, erasing them afterward.
+        options = (
+            self._next_timer_options if self._next_timer_options else _TimerOptions()
+        )
+        self._next_timer_options = None
+        return self._timer_impl(delay, options, callback, *args, context=context)
 
     def call_at(
         self,

temporalio/workflow.py

@@ -224,6 +224,7 @@ def signal(
 def signal(
     *,
     unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,
+    description: Optional[str] = None,
 ) -> Callable[
     [CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType
 ]: ...
@@ -251,16 +252,6 @@ def signal(
 ]: ...
 
 
-@overload
-def signal(
-    *,
-    description: str,
-    unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,
-) -> Callable[
-    [CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType
-]: ...
-
-
 def signal(
     fn: Optional[CallableSyncOrAsyncReturnNoneType] = None,
     *,
@@ -768,7 +759,7 @@ async def workflow_wait_condition(
     def workflow_get_current_details(self) -> str: ...
 
     @abstractmethod
-    def workflow_set_current_details(self, description): ...
+    def workflow_set_current_details(self, details: str): ...
 
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(

tests/worker/test_workflow.py

@@ -6287,3 +6287,21 @@ async def waiting() -> bool:
         describe_r = await handle.describe()
         assert describe_r.static_summary == "cool workflow bro"
         assert describe_r.static_details == "xtremely detailed"
+
+
+@workflow.defn
+class WorkflowSleepWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        await workflow.sleep(1)
+
+
+async def test_workflow_sleep(client: Client):
+    async with new_worker(client, WorkflowSleepWorkflow) as worker:
+        start_time = datetime.now()
+        await client.execute_workflow(
+            WorkflowSleepWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert (datetime.now() - start_time) >= timedelta(seconds=1)