Implement

Author: dandavison

temporalio/activity.py

@@ -36,6 +36,7 @@
 
 import temporalio.common
 import temporalio.converter
+from temporalio.client import Client
 
 from .types import CallableType
 
@@ -147,6 +148,7 @@ class _Context:
         temporalio.converter.PayloadConverter,
     ]
     runtime_metric_meter: Optional[temporalio.common.MetricMeter]
+    client: Client
     _logger_details: Optional[Mapping[str, Any]] = None
     _payload_converter: Optional[temporalio.converter.PayloadConverter] = None
     _metric_meter: Optional[temporalio.common.MetricMeter] = None
@@ -238,13 +240,25 @@ def wait_sync(self, timeout: Optional[float] = None) -> None:
         self.thread_event.wait(timeout)
 
 
+def client() -> Client:
+    """Return a Temporal Client for use in the current activity.
+
+    Returns:
+        :py:class:`temporalio.client.Client` for use in the current activity.
+
+    Raises:
+        RuntimeError: When not in an activity.
+    """
+    return _Context.current().client
+
+
 def in_activity() -> bool:
     """Whether the current code is inside an activity.
 
     Returns:
         True if in an activity, False otherwise.
     """
-    return not _current_context.get(None) is None
+    return _current_context.get(None) is not None
 
 
 def info() -> Info:

temporalio/worker/_activity.py

@@ -69,6 +69,7 @@ def __init__(
         data_converter: temporalio.converter.DataConverter,
         interceptors: Sequence[Interceptor],
         metric_meter: temporalio.common.MetricMeter,
+        client: temporalio.client.Client,
     ) -> None:
         self._bridge_worker = bridge_worker
         self._task_queue = task_queue
@@ -84,6 +85,7 @@ def __init__(
             None
         )
         self._seen_sync_activity = False
+        self._client = client
 
         # Validate and build activity dict
         self._activities: Dict[str, temporalio.activity._Definition] = {}
@@ -428,13 +430,16 @@ async def _run_activity(
                     heartbeat=None,
                     cancelled_event=running_activity.cancelled_event,
                     worker_shutdown_event=self._worker_shutdown_event,
-                    shield_thread_cancel_exception=None
-                    if not running_activity.cancel_thread_raiser
-                    else running_activity.cancel_thread_raiser.shielded,
+                    shield_thread_cancel_exception=(
+                        None
+                        if not running_activity.cancel_thread_raiser
+                        else running_activity.cancel_thread_raiser.shielded
+                    ),
                     payload_converter_class_or_instance=self._data_converter.payload_converter,
-                    runtime_metric_meter=None
-                    if sync_non_threaded
-                    else self._metric_meter,
+                    runtime_metric_meter=(
+                        None if sync_non_threaded else self._metric_meter
+                    ),
+                    client=self._client,
                 )
             )
             temporalio.activity.logger.debug("Starting activity")
@@ -692,6 +697,7 @@ async def heartbeat_with_context(*details: Any) -> None:
                     worker_shutdown_event.thread_event,
                     payload_converter_class_or_instance,
                     ctx.runtime_metric_meter,
+                    ctx.client,
                     input.fn,
                     *input.args,
                 ]
@@ -739,6 +745,7 @@ def _execute_sync_activity(
         temporalio.converter.PayloadConverter,
     ],
     runtime_metric_meter: Optional[temporalio.common.MetricMeter],
+    client: temporalio.client.Client,
     fn: Callable[..., Any],
     *args: Any,
 ) -> Any:
@@ -770,6 +777,7 @@ def _execute_sync_activity(
             else cancel_thread_raiser.shielded,
             payload_converter_class_or_instance=payload_converter_class_or_instance,
             runtime_metric_meter=runtime_metric_meter,
+            client=client,
         )
     )
     return fn(*args)

temporalio/worker/_worker.py

@@ -29,7 +29,7 @@
 
 from ._activity import SharedStateManager, _ActivityWorker
 from ._interceptor import Interceptor
-from ._tuning import WorkerTuner, _to_bridge_slot_supplier
+from ._tuning import WorkerTuner
 from ._workflow import _WorkflowWorker
 from ._workflow_instance import UnsandboxedWorkflowRunner, WorkflowRunner
 from .workflow_sandbox import SandboxedWorkflowRunner
@@ -304,6 +304,7 @@ def __init__(
                 data_converter=client_config["data_converter"],
                 interceptors=interceptors,
                 metric_meter=self._runtime.metric_meter,
+                client=client,
             )
         self._workflow_worker: Optional[_WorkflowWorker] = None
         if workflows: