diff --git a/CHANGELOG.md b/CHANGELOG.md index 20ddd464f..32d347e61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,12 +20,20 @@ to include examples, links to docs, or any other relevant information. ### Added +- Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome. +- New properties and methods in ActivityExecution and ActivityExecutionDescription. + ### Changed ### Deprecated ### :boom: Breaking Changes +- client.ActivityExecution and client.ActivityExecutionDescription had some fields removed or renamed + to match RPC API. Dataclass equality method has been disabled for these types. +- ActivityHandle.describe() long-poll token was removed. The functionality can still be used manually + through raw gRPC API. + ### Fixed ### Security diff --git a/temporalio/client/_activity.py b/temporalio/client/_activity.py index 138d09dc7..ca648f6d7 100644 --- a/temporalio/client/_activity.py +++ b/temporalio/client/_activity.py @@ -9,14 +9,13 @@ Mapping, Sequence, ) -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from enum import IntEnum from typing import ( TYPE_CHECKING, Any, Generic, - cast, ) from typing_extensions import Self @@ -163,7 +162,7 @@ async def __anext__(self) -> ActivityExecution: return ret -@dataclass(frozen=True) +@dataclass(eq=False, kw_only=True) class ActivityExecution: """Info for an activity execution not started by a workflow, from list response. @@ -184,23 +183,17 @@ class ActivityExecution: """Time the activity reached a terminal status, if closed.""" execution_duration: timedelta | None - """Duration from scheduled to close time, only populated if closed.""" + """Duration from schedule to close time, only populated if closed.""" + + execution_time: datetime | None + """The time at which the first activity task is made available for dispatch, computed as schedule time + start delay.""" namespace: str """Namespace of the activity (copied from calling client).""" - raw_info: ( - temporalio.api.activity.v1.ActivityExecutionListInfo - | temporalio.api.activity.v1.ActivityExecutionInfo - ) - """Underlying protobuf info.""" - - scheduled_time: datetime + schedule_time: datetime """Time the activity was originally scheduled.""" - state_transition_count: int | None - """Number of state transitions, if available.""" - status: ActivityExecutionStatus """Current status of the activity.""" @@ -210,17 +203,28 @@ class ActivityExecution: typed_search_attributes: temporalio.common.TypedSearchAttributes """Current set of search attributes if any.""" + raw_info: ( + temporalio.api.activity.v1.ActivityExecutionListInfo + | temporalio.api.activity.v1.ActivityExecutionInfo + ) = field(repr=False) + """Underlying protobuf info.""" + @classmethod def _from_raw_info( - cls, info: temporalio.api.activity.v1.ActivityExecutionListInfo, namespace: str + cls, + info: ( + temporalio.api.activity.v1.ActivityExecutionListInfo + | temporalio.api.activity.v1.ActivityExecutionInfo + ), + namespace: str, + **kwargs: Any, ) -> Self: """Create from raw proto activity list info.""" return cls( + raw_info=info, activity_id=info.activity_id, activity_run_id=info.run_id or None, - activity_type=( - info.activity_type.name if info.HasField("activity_type") else "" - ), + activity_type=info.activity_type.name, close_time=( info.close_time.ToDatetime().replace(tzinfo=timezone.utc) if info.HasField("close_time") @@ -231,16 +235,17 @@ def _from_raw_info( if info.HasField("execution_duration") else None ), + execution_time=( + info.execution_time.ToDatetime().replace(tzinfo=timezone.utc) + if info.HasField("execution_time") + else None + ), namespace=namespace, - raw_info=info, - scheduled_time=( + schedule_time=( info.schedule_time.ToDatetime().replace(tzinfo=timezone.utc) if info.HasField("schedule_time") else datetime.min ), - state_transition_count=( - info.state_transition_count if info.state_transition_count else None - ), status=( ActivityExecutionStatus(info.status) if info.status @@ -250,10 +255,11 @@ def _from_raw_info( typed_search_attributes=temporalio.converter.decode_typed_search_attributes( info.search_attributes ), + **kwargs, ) -@dataclass(frozen=True) +@dataclass(eq=False, kw_only=True) class ActivityExecutionDescription(ActivityExecution): """Detailed information about an activity execution not started by a workflow. @@ -267,20 +273,23 @@ class ActivityExecutionDescription(ActivityExecution): canceled_reason: str | None """Reason for cancellation, if cancel was requested.""" + close_time: datetime | None + """Time when the activity transitioned to a closed state.""" + current_retry_interval: timedelta | None """Time until the next retry, if applicable.""" - eager_execution_requested: bool - """Whether eager execution was requested for this activity.""" + expiration_time: datetime | None + """The time at which the activity's schedule-to-close timeout expires.""" - expiration_time: datetime - """Scheduled time plus schedule_to_close_timeout.""" + heartbeat_timeout: timedelta | None + """Configured heartbeat timeout of the activity.""" last_attempt_complete_time: datetime | None """Time when the last attempt completed.""" - last_failure: Exception | None - """Failure from the last failed attempt, if any.""" + last_deployment_version: temporalio.common.WorkerDeploymentVersion | None + """The Worker Deployment Version this activity was dispatched to most recently.""" last_heartbeat_time: datetime | None """Time of the last heartbeat.""" @@ -288,17 +297,14 @@ class ActivityExecutionDescription(ActivityExecution): last_started_time: datetime | None """Time the last attempt was started.""" - last_worker_identity: str + last_worker_identity: str | None """Identity of the last worker that processed the activity.""" next_attempt_schedule_time: datetime | None """Time when the next attempt will be scheduled.""" - paused: bool - """Whether the activity is paused.""" - - raw_heartbeat_details: Sequence[temporalio.api.common.v1.Payload] - """Details from the last heartbeat.""" + priority: temporalio.common.Priority + """Priority metadata.""" retry_policy: temporalio.common.RetryPolicy | None """Retry policy for the activity.""" @@ -306,114 +312,247 @@ class ActivityExecutionDescription(ActivityExecution): run_state: PendingActivityState | None """More detailed breakdown if status is RUNNING.""" - long_poll_token: bytes | None - """Token for follow-on long-poll requests. None if the activity is complete.""" + schedule_to_close_timeout: timedelta | None + """Configured schedule-to-close timeout of the activity.""" + + schedule_to_start_timeout: timedelta | None + """Configured schedule-to-start timeout of the activity.""" + + start_to_close_timeout: timedelta | None + """Configured start-to-close timeout of the activity.""" + + start_delay: timedelta | None + """Time to wait before making the first activity task available for dispatch.""" - raw_callbacks: Sequence[temporalio.api.activity.v1.CallbackInfo] + total_heartbeat_count: int + """Total number of heartbeats recorded across all attempts of this activity, including retries. + + Zero if the activity has not sent any heartbeats or if the server didn't report heartbeat count. + """ + + raw_info: temporalio.api.activity.v1.ActivityExecutionInfo = field(repr=False) # type: ignore[reportIncompatibleVariableOverride] + """Underlying protobuf info.""" + + raw_callbacks: Sequence[temporalio.api.activity.v1.CallbackInfo] = field(repr=False) """Underlying protobuf callbacks""" + raw_input: temporalio.api.common.v1.Payloads | None = field(repr=False) + """Raw input of the activity. Use :py:meth:`input` to decode.""" + + raw_outcome: temporalio.api.activity.v1.ActivityExecutionOutcome | None = field( + repr=False + ) + """Raw outcome of the activity. Use :py:meth:`outcome` to decode.""" + + data_converter: DataConverter = field(repr=False) + """Data converter used to convert raw payloads. By default it's the same as the client's data converter.""" + @classmethod - async def _from_execution_info( + def _from_resp( cls, - info: temporalio.api.activity.v1.ActivityExecutionInfo, - long_poll_token: bytes | None, + resp: temporalio.api.workflowservice.v1.DescribeActivityExecutionResponse, namespace: str, data_converter: temporalio.converter.DataConverter, - callbacks: Sequence[temporalio.api.activity.v1.CallbackInfo], + **kwargs: Any, ) -> Self: """Create from raw proto activity execution info.""" - # Decode heartbeat details if present - decoded_heartbeat_details: Sequence[temporalio.api.common.v1.Payload] = ( - info.heartbeat_details.payloads - ) - if decoded_heartbeat_details and data_converter.payload_codec: - decoded_heartbeat_details = await data_converter.payload_codec.decode( - decoded_heartbeat_details - ) - - return cls( - activity_id=info.activity_id, - activity_run_id=info.run_id or None, - activity_type=( - info.activity_type.name if info.HasField("activity_type") else "" - ), - attempt=info.attempt, - canceled_reason=info.canceled_reason or None, - close_time=( - info.close_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("close_time") - else None - ), + return cls._from_raw_info( + info=resp.info, + namespace=namespace, + attempt=resp.info.attempt, + canceled_reason=resp.info.canceled_reason or None, current_retry_interval=( - info.current_retry_interval.ToTimedelta() - if info.HasField("current_retry_interval") - else None - ), - eager_execution_requested=getattr(info, "eager_execution_requested", False), - execution_duration=( - info.execution_duration.ToTimedelta() - if info.HasField("execution_duration") + resp.info.current_retry_interval.ToTimedelta() + if resp.info.HasField("current_retry_interval") else None ), expiration_time=( - info.expiration_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("expiration_time") + resp.info.expiration_time.ToDatetime(tzinfo=timezone.utc) + if resp.info.HasField("expiration_time") else datetime.min ), + heartbeat_timeout=( + resp.info.heartbeat_timeout.ToTimedelta() + if resp.info.HasField("heartbeat_timeout") + else None + ), last_attempt_complete_time=( - info.last_attempt_complete_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("last_attempt_complete_time") + resp.info.last_attempt_complete_time.ToDatetime(tzinfo=timezone.utc) + if resp.info.HasField("last_attempt_complete_time") else None ), - last_failure=( - cast( - Exception | None, - await data_converter.decode_failure(info.last_failure), + last_deployment_version=( + temporalio.common.WorkerDeploymentVersion( + deployment_name=resp.info.last_deployment_version.deployment_name, + build_id=resp.info.last_deployment_version.build_id, ) - if info.HasField("last_failure") + if resp.info.HasField("last_deployment_version") else None ), last_heartbeat_time=( - info.last_heartbeat_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("last_heartbeat_time") + resp.info.last_heartbeat_time.ToDatetime(tzinfo=timezone.utc) + if resp.info.HasField("last_heartbeat_time") else None ), last_started_time=( - info.last_started_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("last_started_time") + resp.info.last_started_time.ToDatetime(tzinfo=timezone.utc) + if resp.info.HasField("last_started_time") else None ), - last_worker_identity=info.last_worker_identity, - long_poll_token=long_poll_token or None, - namespace=namespace, + last_worker_identity=resp.info.last_worker_identity or None, next_attempt_schedule_time=( - info.next_attempt_schedule_time.ToDatetime(tzinfo=timezone.utc) - if info.HasField("next_attempt_schedule_time") + resp.info.next_attempt_schedule_time.ToDatetime(tzinfo=timezone.utc) + if resp.info.HasField("next_attempt_schedule_time") + else None + ), + priority=temporalio.common.Priority._from_proto(resp.info.priority), + retry_policy=( + temporalio.common.RetryPolicy.from_proto(resp.info.retry_policy) + if resp.info.HasField("retry_policy") else None ), - paused=getattr(info, "paused", False), - raw_heartbeat_details=decoded_heartbeat_details, - raw_info=info, - retry_policy=temporalio.common.RetryPolicy.from_proto(info.retry_policy) - if info.HasField("retry_policy") - else None, run_state=( - PendingActivityState(info.run_state) if info.run_state else None + PendingActivityState(resp.info.run_state) + if resp.info.run_state + else None ), - scheduled_time=(info.schedule_time.ToDatetime(tzinfo=timezone.utc)), - state_transition_count=( - info.state_transition_count if info.state_transition_count else None + schedule_to_close_timeout=( + resp.info.schedule_to_close_timeout.ToTimedelta() + if resp.info.HasField("schedule_to_close_timeout") + else None ), - status=( - ActivityExecutionStatus(info.status) - if info.status - else ActivityExecutionStatus.UNSPECIFIED + schedule_to_start_timeout=( + resp.info.schedule_to_start_timeout.ToTimedelta() + if resp.info.HasField("schedule_to_start_timeout") + else None ), - task_queue=info.task_queue, - typed_search_attributes=temporalio.converter.decode_typed_search_attributes( - info.search_attributes + start_to_close_timeout=( + resp.info.start_to_close_timeout.ToTimedelta() + if resp.info.HasField("start_to_close_timeout") + else None + ), + start_delay=( + resp.info.start_delay.ToTimedelta() + if resp.info.HasField("start_delay") + else None ), - raw_callbacks=callbacks, + total_heartbeat_count=resp.info.total_heartbeat_count, + raw_callbacks=resp.callbacks, + raw_input=resp.input if resp.HasField("input") else None, + raw_outcome=resp.outcome if resp.HasField("outcome") else None, + data_converter=data_converter, + **kwargs, + ) + + def has_heartbeat_details(self) -> bool: + """True if heartbeat details are available. Use :py:meth:`heartbeat_details` to retrieve them. + + Always false if `include_heartbeat_details` was false in the `describe` call. + """ + return self.raw_info.HasField("heartbeat_details") + + async def heartbeat_details( + self, type_hints: list[type] | None = None + ) -> list[Any] | None: + """Returns details from the last heartbeat, or `None` if not available. + + Always `None` if `include_heartbeat_details` was false in the `describe` call. + Type hints can be provided to aid data conversion. + """ + return ( + await self.data_converter.decode_wrapper( + self.raw_info.heartbeat_details, type_hints + ) + if self.has_heartbeat_details() + else None + ) + + def has_last_failure(self) -> bool: + """True if last failure is available. Use :py:meth:`last_failure` to retrieve it. + + Always false if `include_heartbeat_details` was false in the `describe` call. + """ + return self.raw_info.HasField("last_failure") + + async def last_failure(self) -> BaseException | None: + """Returns failure from the last failed attempt, or `None` if not available. + + Always `None` if `include_last_failure` was false in the `describe` call. + """ + return ( + await self.data_converter.decode_failure(self.raw_info.last_failure) + if self.has_last_failure() + else None + ) + + def has_input(self) -> bool: + """True if activity input is available. Use :py:meth:`input` to retrieve it. + + Always false if `include_input` was false in the `describe` call. + """ + return self.raw_input is not None + + async def input(self, type_hints: list[type] | None = None) -> list[Any] | None: + """Returns activity input, or `None` if not available. + + Always `None` if `include_input` was false in the `describe` call. + Type hints can be provided to aid data conversion. + """ + return ( + await self.data_converter.decode_wrapper(self.raw_input, type_hints) + if self.has_input() + else None + ) + + def has_result(self) -> bool: + """True if activity result is available. Use :py:meth:`result` to retrieve it. + + Activity result is only available if the activity has completed and was successful. + Always false if `include_outcome` was false in the `describe` call. + """ + return self.raw_outcome is not None and self.raw_outcome.HasField("result") + + async def result(self, type_hint: type | None = None) -> Any | None: + """Returns activity result, or `None` if not available. + + Activity result is only available if the activity has completed successfully. + Always false if `include_outcome` was false in the `describe` call. + Type hints can be provided to aid data conversion. + """ + if self.raw_outcome is None or not self.raw_outcome.HasField("result"): + return None + type_hints = [type_hint] if type_hint is not None else None + results = await self.data_converter.decode_wrapper( + self.raw_outcome.result, type_hints + ) + if not results: + return None + if len(results) > 1: + warnings.warn(f"Expected single activity result, got {len(results)}") + return results[0] + + def has_outcome_failure(self) -> bool: + """True if activity outcome failure is available. Use :py:meth:`outcome_failure` to retrieve it. + + Activity outcome failure is only available if the activity has closed with a failure. + Use :py:meth:`last_failure` to retrieve failure of the most recent failed attempt of an activity that's still + running or that completed successfully. + Always false if `include_outcome` was false in the `describe` call. + """ + return self.raw_outcome is not None and self.raw_outcome.HasField("failure") + + async def outcome_failure(self) -> BaseException | None: + """Returns activity outcome failure, or `None` if not available. + + Activity outcome failure is only available if the activity has closed with a failure. + Use :py:meth:`last_failure` to retrieve failure of the most recent failed attempt of an activity that's still + running or that completed successfully. + Always false if `include_outcome` was false in the `describe` call. + """ + return ( + await self.data_converter.decode_failure(self.raw_outcome.failure) + if self.raw_outcome is not None and self.raw_outcome.HasField("failure") + else None ) @@ -889,7 +1028,10 @@ async def terminate( async def describe( self, *, - long_poll_token: bytes | None = None, + include_input: bool = False, + include_outcome: bool = False, + include_heartbeat_details: bool = False, + include_last_failure: bool = False, rpc_metadata: Mapping[str, str | bytes] = {}, rpc_timeout: timedelta | None = None, ) -> ActivityExecutionDescription: @@ -899,8 +1041,10 @@ async def describe( This API is experimental. Args: - long_poll_token: Token from a previous describe response. If provided, - the request will long-poll until the activity state changes. + include_input: Include activity input in the response if available. + include_outcome: Include activity outcome in the response if available. + include_heartbeat_details: Include heartbeat details in the response if available. + include_last_failure: Include last failure in the response if available. rpc_metadata: Headers used on the RPC call. rpc_timeout: Optional RPC deadline to set for the RPC call. @@ -911,7 +1055,10 @@ async def describe( DescribeActivityInput( activity_id=self._id, activity_run_id=self._run_id, - long_poll_token=long_poll_token, + include_input=include_input, + include_outcome=include_outcome, + include_heartbeat_details=include_heartbeat_details, + include_last_failure=include_last_failure, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout, ) diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 78471baf7..08126348d 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -690,15 +690,27 @@ async def describe_activity( namespace=self._client.namespace, activity_id=input.activity_id, run_id=input.activity_run_id or "", - long_poll_token=input.long_poll_token or b"", + include_input=input.include_input, + include_outcome=input.include_outcome, + include_heartbeat_details=input.include_heartbeat_details, + include_last_failure=input.include_last_failure, ), retry=True, metadata=input.rpc_metadata, timeout=input.rpc_timeout, ) - return await ActivityExecutionDescription._from_execution_info( - info=resp.info, - long_poll_token=resp.long_poll_token or None, + # Erase unrequested fields if server sent them anyway (can happen with old server) + if not input.include_input: + resp.ClearField("input") + if not input.include_outcome: + resp.ClearField("outcome") + if not input.include_heartbeat_details: + resp.info.ClearField("heartbeat_details") + if not input.include_last_failure: + resp.info.ClearField("last_failure") + + return ActivityExecutionDescription._from_resp( + resp=resp, namespace=self._client.namespace, data_converter=self._client.data_converter.with_context( ActivitySerializationContext( @@ -711,7 +723,6 @@ async def describe_activity( is_local=False, ) ), - callbacks=resp.callbacks, ) def list_activities( diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index a0310ac19..449d0004c 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -267,7 +267,10 @@ class DescribeActivityInput: activity_id: str activity_run_id: str | None - long_poll_token: bytes | None + include_input: bool + include_outcome: bool + include_heartbeat_details: bool + include_last_failure: bool rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None diff --git a/tests/test_activity.py b/tests/test_activity.py index 3a63f51ac..a5682f221 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -26,6 +26,7 @@ StartActivityInput, TerminateActivityInput, ) +from temporalio.common import RetryPolicy from temporalio.exceptions import ApplicationError, CancelledError from temporalio.service import RPCError, RPCStatusCode from temporalio.testing import WorkflowEnvironment @@ -38,6 +39,16 @@ async def increment(input: int) -> int: return input + 1 +# Activity for testing describe, stores last failure and heartbeat details +@activity.defn +async def heartbeat_fail_increment(x: int) -> int: + if activity.info().heartbeat_details: + return x + 1 + else: + activity.heartbeat("heartbeat details") + raise ApplicationError("first attempt failed") + + # Activity classes for testing start_activity_class / execute_activity_class @activity.defn class IncrementClass: @@ -75,10 +86,6 @@ async def async_increment(self, x: int) -> int: async def async_no_param(self) -> str: return "async-method-result" - @activity.defn - def sync_increment(self, x: int) -> int: - return x + 1 - async def test_start_activity_generates_request_id() -> None: start_activity_execution = mock.AsyncMock( @@ -107,70 +114,118 @@ async def test_start_activity_generates_request_id() -> None: class TestDescribe: @pytest.fixture - async def activity_handle(self, client: Client, env: WorkflowEnvironment): - if env.supports_time_skipping: - pytest.skip( - "Java test server: https://github.com/temporalio/sdk-java/issues/2741" - ) + def task_queue(self): + return str(uuid.uuid4()) + async def make_activity_handle( + self, client: Client, task_queue: str, max_attempts: int = 2 + ): id = str(uuid.uuid4()) - task_queue = str(uuid.uuid4()) - yield await client.start_activity( - increment, - args=(42,), + return await client.start_activity( + heartbeat_fail_increment, + args=(1,), id=id, task_queue=task_queue, - schedule_to_close_timeout=timedelta(hours=1), + schedule_to_close_timeout=timedelta(minutes=1), + retry_policy=RetryPolicy(maximum_attempts=max_attempts), ) + @pytest.fixture + async def activity_handle( + self, client: Client, env: WorkflowEnvironment, task_queue: str + ): + if env.supports_time_skipping: + pytest.skip( + "Java test server: https://github.com/temporalio/sdk-java/issues/2741" + ) + return await self.make_activity_handle(client, task_queue) + async def test_describe(self, client: Client, activity_handle: ActivityHandle): desc = await activity_handle.describe() # From ActivityExecution (base class) assert desc.activity_id == activity_handle.id assert desc.activity_run_id == activity_handle.run_id - assert desc.activity_type == "increment" + assert desc.activity_type == "heartbeat_fail_increment" assert desc.close_time is None # not closed yet assert desc.execution_duration is None # not closed yet assert desc.namespace == client.namespace assert desc.raw_info is not None - assert desc.scheduled_time is not None + assert desc.schedule_time is not None assert len(desc.typed_search_attributes) == 0 - assert desc.state_transition_count is not None assert desc.status == ActivityExecutionStatus.RUNNING assert desc.task_queue # From ActivityExecutionDescription assert desc.attempt == 1 assert desc.canceled_reason is None assert desc.current_retry_interval is None - assert desc.eager_execution_requested is False assert desc.expiration_time is not None - assert len(desc.raw_heartbeat_details) == 0 assert desc.run_state == PendingActivityState.SCHEDULED assert desc.last_attempt_complete_time is None - assert desc.last_failure is None assert desc.last_heartbeat_time is None assert desc.last_started_time is None - assert desc.last_worker_identity == "" - assert desc.long_poll_token is not None + assert desc.last_worker_identity is None assert desc.next_attempt_schedule_time is None - assert desc.paused is False assert desc.retry_policy is not None - async def test_describe_long_poll(self, activity_handle: ActivityHandle): - desc1 = await activity_handle.describe() - assert desc1.long_poll_token - desc2_task = asyncio.create_task( - activity_handle.describe(long_poll_token=desc1.long_poll_token) - ) - # Worker poll causes a transition to Started which notifies the waiting long-poll. + async def test_describe_payloads( + self, client: Client, activity_handle: ActivityHandle, task_queue: str + ): async with Worker( - activity_handle._client, - task_queue=desc1.task_queue, - activities=[increment], + client, + task_queue=task_queue, + activities=[heartbeat_fail_increment], ): - desc2 = await desc2_task - assert desc2.state_transition_count and desc1.state_transition_count - assert desc2.state_transition_count > desc1.state_transition_count + assert await activity_handle.result() == 2 + + desc_default = await activity_handle.describe() + assert not desc_default.has_input() + assert not desc_default.has_result() + assert not desc_default.has_outcome_failure() + assert not desc_default.has_heartbeat_details() + assert not desc_default.has_last_failure() + assert await desc_default.input([int]) is None + assert await desc_default.result(int) is None + assert await desc_default.outcome_failure() is None + assert await desc_default.heartbeat_details([str]) is None + assert await desc_default.last_failure() is None + + desc_payloads = await activity_handle.describe( + include_input=True, + include_outcome=True, + include_heartbeat_details=True, + include_last_failure=True, + ) + assert desc_payloads.has_input() + assert desc_payloads.has_result() + assert not desc_payloads.has_outcome_failure() + assert desc_payloads.has_heartbeat_details() + assert desc_payloads.has_last_failure() + assert await desc_payloads.input([int]) == [1] + assert await desc_payloads.result(int) == 2 + assert await desc_payloads.outcome_failure() is None + assert await desc_payloads.heartbeat_details([str]) == ["heartbeat details"] + assert isinstance(await desc_payloads.last_failure(), ApplicationError) + + failed_activity_handle = await self.make_activity_handle( + client, task_queue, max_attempts=1 + ) + with pytest.raises(ActivityFailureError) as err: + await failed_activity_handle.result() + assert isinstance(err.value.cause, ApplicationError) + + desc_failure = await failed_activity_handle.describe( + include_outcome=True, include_last_failure=True + ) + assert not desc_failure.has_input() + assert not desc_failure.has_result() + assert desc_failure.has_outcome_failure() + assert not desc_failure.has_heartbeat_details() + assert desc_failure.has_last_failure() + assert await desc_failure.input([int]) is None + assert await desc_failure.result(int) is None + assert isinstance(await desc_failure.outcome_failure(), ApplicationError) + assert await desc_failure.heartbeat_details([str]) is None + assert isinstance(await desc_failure.last_failure(), ApplicationError) class ActivityTracingInterceptor(Interceptor): @@ -520,7 +575,7 @@ async def test_start_activity_start_delay(client: Client, env: WorkflowEnvironme desc = await activity_handle.describe() assert desc.last_started_time is not None assert ( - desc.last_started_time - desc.scheduled_time + desc.last_started_time - desc.schedule_time ).total_seconds() >= start_delay.total_seconds() - 0.5 @@ -595,9 +650,6 @@ async def check_executions(): assert execution.activity_type == "increment" assert execution.task_queue == task_queue assert execution.status == ActivityExecutionStatus.RUNNING - assert ( - execution.state_transition_count is None - ) # Not set until activity completes await assert_eventually(check_executions)