"""Client support for accessing Temporal."""

from __future__ import annotations

import asyncio

import functools

import warnings

from collections.abc import (

Mapping,

Sequence,

)

from dataclasses import dataclass, field

from datetime import datetime, timedelta, timezone

from enum import IntEnum

from typing import (

TYPE_CHECKING,

Any,

Generic,

TypeVar,

)

from typing_extensions import Self

import temporalio.api.activity.v1

import temporalio.api.common.v1

import temporalio.api.enums.v1

import temporalio.api.workflowservice.v1

import temporalio.common

import temporalio.converter

import temporalio.converter._search_attributes

from temporalio.converter import (

ActivitySerializationContext,

DataConverter,

SerializationContext,

WithSerializationContext,

)

from temporalio.service import (

RPCError,

RPCStatusCode,

)

from ..types import (

ReturnType,

)

from ._exceptions import ActivityFailureError

from ._interceptor import (

CancelActivityInput,

CompleteAsyncActivityInput,

DescribeActivityInput,

FailAsyncActivityInput,

HeartbeatAsyncActivityInput,

PauseActivityInput,

ReportCancellationAsyncActivityInput,

TerminateActivityInput,

UnpauseActivityInput,

UpdateActivityOptionsInput,

)

if TYPE_CHECKING:

from ._client import Client

from ._interceptor import ListActivitiesInput

class ActivityExecutionAsyncIterator:

"""Asynchronous iterator for activity execution values.

You should typically use ``async for`` on this iterator and not call any of its methods.

"""

def __init__(

self,

client: Client,

input: ListActivitiesInput,

) -> None:

"""Create an asynchronous iterator for the given input.

Users should not create this directly, but rather use

:py:meth:`Client.list_activities`.

"""

self._client = client

self._input = input

self._next_page_token = input.next_page_token

self._current_page: Sequence[ActivityExecution] | None = None

self._current_page_index = 0

self._limit = input.limit

self._yielded = 0

@property

def current_page_index(self) -> int:

"""Index of the entry in the current page that will be returned from

the next :py:meth:`__anext__` call.

"""

return self._current_page_index

@property

def current_page(self) -> Sequence[ActivityExecution] | None:

"""Current page, if it has been fetched yet."""

return self._current_page

@property

def next_page_token(self) -> bytes | None:

"""Token for the next page request if any."""

return self._next_page_token

async def fetch_next_page(self, *, page_size: int | None = None) -> None:

"""Fetch the next page of results.

Args:

page_size: Override the page size this iterator was originally

created with.

"""

page_size = page_size or self._input.page_size

if self._limit is not None and self._limit - self._yielded < page_size:

page_size = self._limit - self._yielded

resp = await self._client.workflow_service.list_activity_executions(

temporalio.api.workflowservice.v1.ListActivityExecutionsRequest(

namespace=self._client.namespace,

page_size=page_size,

next_page_token=self._next_page_token or b"",

query=self._input.query or "",

),

retry=True,

metadata=self._input.rpc_metadata,

timeout=self._input.rpc_timeout,

)

self._current_page = [

ActivityExecution._from_raw_info(v, self._client.namespace)

for v in resp.executions

]

self._current_page_index = 0

self._next_page_token = resp.next_page_token or None

def __aiter__(self) -> ActivityExecutionAsyncIterator:

"""Return self as the iterator."""

return self

async def __anext__(self) -> ActivityExecution:

"""Get the next execution on this iterator, fetching next page if

necessary.

"""

if self._limit is not None and self._yielded >= self._limit:

raise StopAsyncIteration

while True:

# No page? fetch and continue

if self._current_page is None:

await self.fetch_next_page()

continue

# No more left in page?

if self._current_page_index >= len(self._current_page):

# If there is a next page token, try to get another page and try

# again

if self._next_page_token is not None:

await self.fetch_next_page()

continue

# No more pages means we're done

raise StopAsyncIteration

# Get current, increment page index, and return

ret = self._current_page[self._current_page_index]

self._current_page_index += 1

self._yielded += 1

return ret

@dataclass(frozen=True, eq=False, kw_only=True)

class ActivityExecution:

"""Info for an activity execution not started by a workflow, from list response."""

activity_id: str

"""Activity ID."""

activity_run_id: str | None

"""Run ID of the activity."""

activity_type: str

"""Type name of the activity."""

close_time: datetime | None

"""Time the activity reached a terminal status, if closed."""

execution_duration: timedelta | None

"""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)."""

schedule_time: datetime

"""Time the activity was originally scheduled."""

status: ActivityExecutionStatus

"""Current status of the activity."""

task_queue: str

"""Task queue the activity was scheduled on."""

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

| 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,

close_time=(

info.close_time.ToDatetime().replace(tzinfo=timezone.utc)

if info.HasField("close_time")

else None

),

execution_duration=(

info.execution_duration.ToTimedelta()

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,

schedule_time=(

info.schedule_time.ToDatetime().replace(tzinfo=timezone.utc)

if info.HasField("schedule_time")

else datetime.min

),

status=(

ActivityExecutionStatus(info.status)

if info.status

else ActivityExecutionStatus.UNSPECIFIED

),

task_queue=info.task_queue,

typed_search_attributes=temporalio.converter.decode_typed_search_attributes(

info.search_attributes

),

**kwargs,

)

@dataclass(frozen=True, eq=False, kw_only=True)

class ActivityExecutionDescription(ActivityExecution):

"""Detailed information about an activity execution not started by a workflow."""

attempt: int

"""Current attempt number."""

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."""

expiration_time: datetime | None

"""The time at which the activity's schedule-to-close timeout expires."""

heartbeat_timeout: timedelta | None

"""Configured heartbeat timeout of the activity."""

last_attempt_complete_time: datetime | None

"""Time when the last attempt completed."""

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."""

last_started_time: datetime | None

"""Time the last attempt was started."""

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."""

priority: temporalio.common.Priority

"""Priority metadata."""

retry_policy: temporalio.common.RetryPolicy | None

"""Retry policy for the activity."""

run_state: PendingActivityState | None

"""More detailed breakdown if status is RUNNING."""

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."""

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)

"""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 `input` to decode."""

raw_outcome: temporalio.api.activity.v1.ActivityExecutionOutcome | None = field(

repr=False

)

"""Raw outcome of the activity. Use :py:meth:`result` or :py:meth:`outcome_failure` 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

def _from_resp(

cls,

resp: temporalio.api.workflowservice.v1.DescribeActivityExecutionResponse,

namespace: str,

data_converter: temporalio.converter.DataConverter,

**kwargs: Any,

) -> Self:

"""Create from raw proto activity execution info."""

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=(

resp.info.current_retry_interval.ToTimedelta()

if resp.info.HasField("current_retry_interval")

else None

),

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=(

resp.info.last_attempt_complete_time.ToDatetime(tzinfo=timezone.utc)

if resp.info.HasField("last_attempt_complete_time")

else None

),

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 resp.info.HasField("last_deployment_version")

else None

),

last_heartbeat_time=(

resp.info.last_heartbeat_time.ToDatetime(tzinfo=timezone.utc)

if resp.info.HasField("last_heartbeat_time")

else None

),

last_started_time=(

resp.info.last_started_time.ToDatetime(tzinfo=timezone.utc)

if resp.info.HasField("last_started_time")

else None

),

last_worker_identity=resp.info.last_worker_identity or None,

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

),

run_state=(

PendingActivityState(resp.info.run_state)

if resp.info.run_state

else None

),

schedule_to_close_timeout=(

resp.info.schedule_to_close_timeout.ToTimedelta()

if resp.info.HasField("schedule_to_close_timeout")

else None

),

schedule_to_start_timeout=(

resp.info.schedule_to_start_timeout.ToTimedelta()

if resp.info.HasField("schedule_to_start_timeout")

else None

),

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

),

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 `heartbeat_details` to retrieve them.

Always false if `include_heartbeat_details` was false in the `ActivityHandle.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 `ActivityHandle.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 `last_failure` to retrieve it.

Always false if ``include_last_failure`` was false in the `ActivityHandle.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 `ActivityHandle.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 `input <ActivityExecutionDescription.input>` to retrieve it.

Always false if ``include_input`` was false in the `ActivityHandle.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 `ActivityHandle.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 `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 `ActivityHandle.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 `ActivityHandle.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 `outcome_failure` to retrieve it.

Activity outcome failure is only available if the activity has closed with a failure.

Use `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 `ActivityHandle.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 `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 `ActivityHandle.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

)

class ActivityExecutionStatus(IntEnum):

"""Status of an activity execution.

See :py:class:`temporalio.api.enums.v1.ActivityExecutionStatus`.

"""

UNSPECIFIED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED

)

RUNNING = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_RUNNING

)

COMPLETED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_COMPLETED

)

FAILED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_FAILED

)

CANCELED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_CANCELED

)

TERMINATED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_TERMINATED

)

TIMED_OUT = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_TIMED_OUT

)

PAUSED = int(

temporalio.api.enums.v1.ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_PAUSED

)

class PendingActivityState(IntEnum):

"""Detailed state of an activity execution that is in ACTIVITY_EXECUTION_STATUS_RUNNING.

See :py:class:`temporalio.api.enums.v1.PendingActivityState`.

"""

UNSPECIFIED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_UNSPECIFIED

)

SCHEDULED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_SCHEDULED

)

STARTED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_STARTED

)

CANCEL_REQUESTED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED

)

PAUSED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED

)

PAUSE_REQUESTED = int(

temporalio.api.enums.v1.PendingActivityState.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED

)

ActivityOptionValueType = TypeVar("ActivityOptionValueType")

@dataclass(frozen=True)

class ActivityOptionsKey(Generic[ActivityOptionValueType]):

"""Typed key for one updatable activity option.

Use the keys on :py:class:`ActivityOptionsKeys` rather than constructing

these directly.

.. warning::

This API is experimental.

"""

name: str

"""Field-mask path this key updates."""

def value_set(

self, value: ActivityOptionValueType

) -> ActivityOptionsUpdate[ActivityOptionValueType]:

"""Create an update that sets this option to the given value."""

return ActivityOptionsUpdate(self, value)

def value_unset(self) -> ActivityOptionsUpdate[ActivityOptionValueType]:

"""Create an update that clears this option server-side."""

return ActivityOptionsUpdate(self, None)

@dataclass(frozen=True)

class ActivityOptionsUpdate(Generic[ActivityOptionValueType]):

"""A single change to an activity's options.

An option not represented by any update in the call is left untouched; an

update carrying None clears the option.

.. warning::

This API is experimental.

"""

key: ActivityOptionsKey[ActivityOptionValueType]

"""Option being changed."""

value: ActivityOptionValueType | None

"""Value being set, or None to clear the option."""

class ActivityOptionsKeys:

"""The activity options that :py:meth:`ActivityHandle.update_options` can change.

.. warning::

This API is experimental.

"""

task_queue: ActivityOptionsKey[str] = ActivityOptionsKey("task_queue.name")

schedule_to_close_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey(

"schedule_to_close_timeout"

)

schedule_to_start_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey(

"schedule_to_start_timeout"

)

start_to_close_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey(

"start_to_close_timeout"

)

heartbeat_timeout: ActivityOptionsKey[timedelta] = ActivityOptionsKey(

"heartbeat_timeout"

)

start_delay: ActivityOptionsKey[timedelta] = ActivityOptionsKey("start_delay")

retry_policy: ActivityOptionsKey[temporalio.common.RetryPolicy] = (

ActivityOptionsKey("retry_policy")

)

priority: ActivityOptionsKey[temporalio.common.Priority] = ActivityOptionsKey(

"priority"

)

@dataclass(frozen=True)

class ActivityExecutionOptions:

"""An activity's options as resolved by the server.

Returned by :py:meth:`ActivityHandle.update_options` and

:py:meth:`ActivityHandle.restore_original_options`.

.. warning::

This API is experimental.

"""

task_queue: str | None

"""Task queue the activity is scheduled on."""

schedule_to_close_timeout: timedelta | None

"""Total time the caller is willing to wait, including retries."""

schedule_to_start_timeout: timedelta | None

"""Maximum time the activity may wait to be picked up by a worker."""

start_to_close_timeout: timedelta | None

"""Maximum time for a single attempt."""

heartbeat_timeout: timedelta | None

"""Maximum allowed time between heartbeats."""

start_delay: timedelta | None

"""Delay before the first attempt is made available for dispatch."""

retry_policy: temporalio.common.RetryPolicy | None

"""Retry policy in effect for the activity."""

priority: temporalio.common.Priority | None

"""Priority of the activity."""

@staticmethod

def _from_proto(

options: temporalio.api.activity.v1.ActivityOptions,

) -> ActivityExecutionOptions:

return ActivityExecutionOptions(

task_queue=options.task_queue.name

if options.HasField("task_queue")

else None,

schedule_to_close_timeout=(

options.schedule_to_close_timeout.ToTimedelta()

if options.HasField("schedule_to_close_timeout")

else None

),

schedule_to_start_timeout=(

options.schedule_to_start_timeout.ToTimedelta()

if options.HasField("schedule_to_start_timeout")

else None

),

start_to_close_timeout=(

options.start_to_close_timeout.ToTimedelta()

if options.HasField("start_to_close_timeout")

else None

),

heartbeat_timeout=(

options.heartbeat_timeout.ToTimedelta()

if options.HasField("heartbeat_timeout")

else None

),

start_delay=(

options.start_delay.ToTimedelta()

if options.HasField("start_delay")

else None

),

retry_policy=(

temporalio.common.RetryPolicy.from_proto(options.retry_policy)

if options.HasField("retry_policy")

else None

),

priority=(

temporalio.common.Priority._from_proto(options.priority)

if options.HasField("priority")

else None

),

)

@dataclass(frozen=True)

class ActivityExecutionCount:

"""Representation of a count from a count activities call."""

count: int

"""Total count matching the filter, if any."""

groups: Sequence[ActivityExecutionCountAggregationGroup]

"""Aggregation groups if requested."""

@staticmethod

def _from_raw(

resp: temporalio.api.workflowservice.v1.CountActivityExecutionsResponse,

) -> ActivityExecutionCount:

"""Create from raw proto response."""

return ActivityExecutionCount(

count=resp.count,

groups=[

ActivityExecutionCountAggregationGroup._from_raw(g) for g in resp.groups

],

)

@dataclass(frozen=True)

class ActivityExecutionCountAggregationGroup:

"""A single aggregation group from a count activities call."""

count: int

"""Count for this group."""

group_values: Sequence[temporalio.common.SearchAttributeValue]

"""Values that define this group."""

@staticmethod

def _from_raw(

raw: temporalio.api.workflowservice.v1.CountActivityExecutionsResponse.AggregationGroup,

) -> ActivityExecutionCountAggregationGroup:

return ActivityExecutionCountAggregationGroup(

count=raw.count,

group_values=[

temporalio.converter._search_attributes._decode_search_attribute_value(

v

)

for v in raw.group_values

],

)

@dataclass(frozen=True)

class AsyncActivityIDReference:

"""Reference to an async activity by its qualified ID."""

workflow_id: str | None

run_id: str | None

activity_id: str

class AsyncActivityHandle(WithSerializationContext):

"""Handle representing an external activity for completion and heartbeat."""

def __init__(

self,

client: Client,

id_or_token: AsyncActivityIDReference | bytes,

data_converter_override: DataConverter | None = None,

) -> None:

"""Create an async activity handle."""

self._client = client

self._id_or_token = id_or_token

self._data_converter_override = data_converter_override

async def heartbeat(

self,

*details: Any,

rpc_metadata: Mapping[str, str | bytes] = {},

rpc_timeout: timedelta | None = None,

) -> None:

"""Record a heartbeat for the activity.

Args:

details: Details of the heartbeat.

rpc_metadata: Headers used on the RPC call. Keys here override

client-level RPC metadata keys.

rpc_timeout: Optional RPC deadline to set for the RPC call.

"""

await self._client._impl.heartbeat_async_activity(

HeartbeatAsyncActivityInput(

id_or_token=self._id_or_token,

details=details,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

data_converter_override=self._data_converter_override,

),

)

async def complete(

self,

result: Any | None = temporalio.common._arg_unset,

*,

rpc_metadata: Mapping[str, str | bytes] = {},

rpc_timeout: timedelta | None = None,

) -> None:

"""Complete the activity.

Args:

result: Result of the activity if any.

rpc_metadata: Headers used on the RPC call. Keys here override

client-level RPC metadata keys.

rpc_timeout: Optional RPC deadline to set for the RPC call.

"""

await self._client._impl.complete_async_activity(

CompleteAsyncActivityInput(

id_or_token=self._id_or_token,

result=result,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

data_converter_override=self._data_converter_override,

),

)

async def fail(

self,

error: Exception,

*,

last_heartbeat_details: Sequence[Any] = [],

rpc_metadata: Mapping[str, str | bytes] = {},

rpc_timeout: timedelta | None = None,

) -> None:

"""Fail the activity.

Args:

error: Error for the activity.

last_heartbeat_details: Last heartbeat details for the activity.

rpc_metadata: Headers used on the RPC call. Keys here override

client-level RPC metadata keys.

rpc_timeout: Optional RPC deadline to set for the RPC call.

"""

await self._client._impl.fail_async_activity(

FailAsyncActivityInput(

id_or_token=self._id_or_token,

error=error,

last_heartbeat_details=last_heartbeat_details,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

data_converter_override=self._data_converter_override,

),

)

async def report_cancellation(

self,

*details: Any,

rpc_metadata: Mapping[str, str | bytes] = {},

rpc_timeout: timedelta | None = None,

) -> None:

"""Report the activity as cancelled.

Args:

details: Cancellation details.

rpc_metadata: Headers used on the RPC call. Keys here override

client-level RPC metadata keys.

rpc_timeout: Optional RPC deadline to set for the RPC call.

"""

await self._client._impl.report_cancellation_async_activity(

ReportCancellationAsyncActivityInput(

id_or_token=self._id_or_token,

details=details,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

data_converter_override=self._data_converter_override,

),

)

def with_context(self, context: SerializationContext) -> Self:

"""Create a new AsyncActivityHandle with a different serialization context.

Payloads received by the activity will be decoded and deserialized using a data converter

with :py:class:`ActivitySerializationContext` set as context. If you are using a custom data

converter that makes use of this context then you can use this method to supply matching

context data to the data converter used to serialize and encode the outbound payloads.

"""

data_converter = self._client.data_converter.with_context(context)

if data_converter is self._client.data_converter:

return self

cls = type(self)

if cls.__init__ is not AsyncActivityHandle.__init__:

raise TypeError(

"If you have subclassed AsyncActivityHandle and overridden the __init__ method "

"then you must override with_context to return an instance of your class."

)

return cls(

self._client,

self._id_or_token,

data_converter,

)

class ActivityHandle(Generic[ReturnType]):

"""Handle representing an activity execution not started by a workflow."""

def __init__(

self,

client: Client,

id: str,

*,

run_id: str | None = None,

result_type: type | None = None,

) -> None:

"""Create activity handle."""

self._client = client

self._id = id

self._run_id = run_id

self._result_type = result_type

self._known_outcome: (

temporalio.api.activity.v1.ActivityExecutionOutcome | None

) = None

@functools.cached_property

def _data_converter(self) -> temporalio.converter.DataConverter:

return self._client.data_converter.with_context(

ActivitySerializationContext(

namespace=self._client.namespace,

activity_id=self._id,

activity_type=None,

activity_task_queue=None,

is_local=False,

workflow_id=None,

workflow_type=None,

)

)

@property

def id(self) -> str:

"""ID of the activity."""

return self._id