Attach everywhere & add test

Author: Sushisource

temporalio/activity.py

@@ -116,6 +116,7 @@ class Info:
     workflow_namespace: str
     workflow_run_id: str
     workflow_type: str
+    priority: temporalio.common.Priority
     # TODO(cretz): Consider putting identity on here for "worker_id" for logger?
 
     def _logger_details(self) -> Mapping[str, Any]:

temporalio/client.py

@@ -2446,6 +2446,7 @@ def __init__(
         start_delay: Optional[timedelta] = None,
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
+        priority: Optional[temporalio.common.Priority] = None,
     ) -> None: ...
 
     # Overload for single-param workflow, with_start
@@ -2508,6 +2509,7 @@ def __init__(
         start_delay: Optional[timedelta] = None,
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
+        priority: Optional[temporalio.common.Priority] = None,
     ) -> None: ...
 
     # Overload for string-name workflow, with_start
@@ -2540,6 +2542,7 @@ def __init__(
         start_delay: Optional[timedelta] = None,
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
+        priority: Optional[temporalio.common.Priority] = None,
     ) -> None: ...
 
     def __init__(
@@ -2570,6 +2573,7 @@ def __init__(
         start_delay: Optional[timedelta] = None,
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
+        priority: Optional[temporalio.common.Priority] = None,
         stack_level: int = 2,
     ) -> None:
         """Create a WithStartWorkflowOperation.
@@ -2610,6 +2614,7 @@ def __init__(
             ret_type=result_type or result_type_from_run_fn,
             rpc_metadata=rpc_metadata,
             rpc_timeout=rpc_timeout,
+            priority=priority,
         )
         self._workflow_handle: Future[WorkflowHandle[SelfType, ReturnType]] = Future()
         self._used = False

temporalio/common.py

@@ -30,7 +30,7 @@
 )
 
 import google.protobuf.internal.containers
-from typing_extensions import ClassVar, NamedTuple, Self, TypeAlias, get_origin
+from typing_extensions import NamedTuple, Self, TypeAlias, get_origin
 
 import temporalio.api.common.v1
 import temporalio.api.enums.v1
@@ -992,6 +992,11 @@ class Priority:
     3.
     """
 
+    @staticmethod
+    def from_proto(proto: temporalio.api.common.v1.Priority) -> Priority:
+        """Create a Priority instance from the proto object."""
+        return Priority(priority_key=proto.priority_key)
+
     def _to_proto(self) -> temporalio.api.common.v1.Priority:
         return temporalio.api.common.v1.Priority(
             priority_key=self.priority_key,

temporalio/testing/_activity.py

@@ -39,6 +39,7 @@
     workflow_namespace="default",
     workflow_run_id="test-run",
     workflow_type="test",
+    priority=temporalio.common.Priority(priority_key=0),
 )
 
 

temporalio/worker/_activity.py

@@ -411,6 +411,7 @@ async def _run_activity(
                 workflow_namespace=start.workflow_namespace,
                 workflow_run_id=start.workflow_execution.run_id,
                 workflow_type=start.workflow_type,
+                priority=temporalio.common.Priority.from_proto(start.priority),
             )
             running_activity.info = info
             input = ExecuteActivityInput(

temporalio/worker/_workflow.py

@@ -404,6 +404,7 @@ def _create_workflow_instance(
             ),
             workflow_id=init.workflow_id,
             workflow_type=init.workflow_type,
+            priority=temporalio.common.Priority.from_proto(init.priority),
         )
 
         # Create instance from details

temporalio/worker/workflow_sandbox/_runner.py

@@ -46,6 +46,7 @@
     typed_search_attributes=temporalio.common.TypedSearchAttributes.empty,
     workflow_id="sandbox-validate-workflow_id",
     workflow_type="sandbox-validate-workflow_type",
+    priority=temporalio.common.Priority(priority_key=0),
 )
 
 

temporalio/workflow.py

@@ -427,6 +427,9 @@ class Info:
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     namespace: str
     parent: Optional[ParentInfo]
+    priority: temporalio.common.Priority
+    """The priority of this workflow execution. If not set, or this server predates priorities,
+    then returns a default instance."""
     raw_memo: Mapping[str, temporalio.api.common.v1.Payload]
     retry_policy: Optional[temporalio.common.RetryPolicy]
     run_id: str

tests/worker/test_workflow.py

@@ -6034,9 +6034,9 @@ async def _do_first_completion_command_is_honored_test(
             result = await handle.result()
         except WorkflowFailureError as err:
             if main_workflow_returns_before_signal_completions:
-                assert (
-                    False
-                ), "Expected no error due to main workflow coroutine returning first"
+                assert False, (
+                    "Expected no error due to main workflow coroutine returning first"
+                )
             else:
                 assert str(err.cause).startswith("Client should see this error")
         else:
@@ -6993,10 +6993,17 @@ async def test_update_handler_semaphore_acquisition_respects_timeout(
     )
 
 
+@activity.defn
+async def check_priority_activity(should_have_priorty: int) -> str:
+    assert activity.info().priority.priority_key == should_have_priorty
+    return "Done!"
+
+
 @workflow.defn
 class WorkflowUsingPriorities:
     @workflow.run
     async def run(self, name: str) -> str:
+        assert workflow.info().priority.priority_key == 1
         await workflow.execute_child_workflow(
             HelloWorkflow.run, name, priority=Priority(priority_key=4)
         )