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

from __future__ import annotations

from collections.abc import (

Awaitable,

Callable,

Mapping,

Sequence,

)

from datetime import timedelta

from typing import (

TYPE_CHECKING,

Any,

Concatenate,

cast,

overload,

)

import nexusrpc

from nexusrpc import OutputT

from typing_extensions import Required, Self, TypedDict

import temporalio.activity

import temporalio.api.workflowservice.v1

import temporalio.common

import temporalio.converter

import temporalio.runtime

import temporalio.service

import temporalio.workflow

from temporalio.service import (

ConnectConfig,

DnsLoadBalancingConfig,

GrpcCompression,

HttpConnectProxyConfig,

KeepAliveConfig,

PayloadLimitsConfig,

RetryConfig,

ServiceClient,

TLSConfig,

)

from ..common import HeaderCodecBehavior

from ..types import (

CallableAsyncNoParam,

CallableAsyncSingleParam,

CallableSyncNoParam,

CallableSyncSingleParam,

LocalReturnType,

MethodAsyncNoParam,

MethodAsyncSingleParam,

MultiParamSpec,

NexusServiceType,

ParamType,

ReturnType,

SelfType,

)

from ._activity import (

ActivityExecutionAsyncIterator,

ActivityExecutionCount,

ActivityHandle,

AsyncActivityHandle,

AsyncActivityIDReference,

)

from ._impl import _ClientImpl

from ._interceptor import (

CountActivitiesInput,

CountNexusOperationsInput,

CountWorkflowsInput,

CreateScheduleInput,

GetWorkerBuildIdCompatibilityInput,

GetWorkerTaskReachabilityInput,

ListActivitiesInput,

ListNexusOperationsInput,

ListSchedulesInput,

ListWorkflowsInput,

OutboundInterceptor,

StartActivityInput,

StartWorkflowInput,

StartWorkflowUpdateWithStartInput,

UpdateWithStartUpdateWorkflowInput,

UpdateWorkerBuildIdCompatibilityInput,

)

from ._nexus import (

NexusClient,

NexusOperationExecutionAsyncIterator,

NexusOperationExecutionCount,

NexusOperationHandle,

_NexusClient,

)

from ._schedule import (

Schedule,

ScheduleAsyncIterator,

ScheduleBackfill,

ScheduleHandle,

)

from ._worker_versioning import (

BuildIdOp,

TaskReachabilityType,

WorkerBuildIdVersionSets,

WorkerTaskReachability,

)

from ._workflow import (

WithStartWorkflowOperation,

WorkflowExecutionAsyncIterator,

WorkflowExecutionCount,

WorkflowHandle,

WorkflowUpdateHandle,

WorkflowUpdateStage,

)

if TYPE_CHECKING:

from ._interceptor import Interceptor

from ._plugin import Plugin

class Client:

"""Client for accessing Temporal.

Most users will use :py:meth:`connect` to create a client. The

:py:attr:`service` property provides access to a raw gRPC client. To create

another client, like for a different namespace, :py:func:`Client` may be

directly instantiated with a :py:attr:`service` of another.

Clients are not thread-safe and should only be used in the event loop they

are first connected in. If a client needs to be used from another thread

than where it was created, make sure the event loop where it was created is

captured, and then call :py:func:`asyncio.run_coroutine_threadsafe` with the

client call and that event loop.

Clients do not work across forks since runtimes do not work across forks.

"""

@classmethod

async def connect(

cls,

target_host: str,

*,

namespace: str = "default",

api_key: str | None = None,

data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default,

plugins: Sequence[Plugin] = [],

interceptors: Sequence[Interceptor] = [],

default_workflow_query_reject_condition: None

| (temporalio.common.QueryRejectCondition) = None,

tls: bool | TLSConfig | None = None,

retry_config: RetryConfig | None = None,

keep_alive_config: KeepAliveConfig | None = KeepAliveConfig.default,

connect_timeout: timedelta | None = None,

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

identity: str | None = None,

lazy: bool = False,

runtime: temporalio.runtime.Runtime | None = None,

http_connect_proxy_config: HttpConnectProxyConfig | None = None,

dns_load_balancing_config: DnsLoadBalancingConfig | None = None,

grpc_compression: GrpcCompression = GrpcCompression.GZIP,

payload_limits: PayloadLimitsConfig = PayloadLimitsConfig(),

header_codec_behavior: HeaderCodecBehavior = HeaderCodecBehavior.NO_CODEC,

) -> Self:

"""Connect to a Temporal server.

Args:

target_host: ``host:port`` for the Temporal server. For local

development, this is often "localhost:7233".

namespace: Namespace to use for client calls.

api_key: API key for Temporal. This becomes the "Authorization"

HTTP header with "Bearer " prepended. This is only set if RPC

metadata doesn't already have an "authorization" key.

data_converter: Data converter to use for all data conversions

to/from payloads.

plugins: Set of plugins that are chained together to allow

intercepting and modifying client creation and service connection.

The earlier plugins wrap the later ones.

Any plugins that also implement

:py:class:`temporalio.worker.Plugin` will be used as worker

plugins too so they should not be given when creating a

worker.

interceptors: Set of interceptors that are chained together to allow

intercepting of client calls. The earlier interceptors wrap the

later ones.

Any interceptors that also implement

:py:class:`temporalio.worker.Interceptor` will be used as worker

interceptors too so they should not be given when creating a

worker.

default_workflow_query_reject_condition: The default rejection

condition for workflow queries if not set during query. See

:py:meth:`WorkflowHandle.query` for details on the rejection

condition.

tls: If ``None``, the default, TLS will be enabled automatically

when ``api_key`` is provided, otherwise TLS is disabled. If

``False``, do not use TLS. If ``True``, use system default TLS

configuration. If TLS configuration present, that TLS

configuration will be used.

retry_config: Retry configuration for direct service calls (when

opted in) or all high-level calls made by this client (which all

opt-in to retries by default). If unset, a default retry

configuration is used.

keep_alive_config: Keep-alive configuration for the client

connection. Default is to check every 30s and kill the

connection if a response doesn't come back in 15s. Can be set to

``None`` to disable.

connect_timeout: Maximum duration of a TCP connection attempt.

If unset, no explicit connection timeout is applied. This is

separate from RPC timeouts and does not bound the total duration

of a call with retries.

rpc_metadata: Headers to use for all calls to the server. Keys here

can be overriden by per-call RPC metadata keys.

identity: Identity for this client. If unset, a default is created

based on the version of the SDK.

lazy: If true, the client will not connect until the first call is

attempted or a worker is created with it. Lazy clients cannot be

used for workers.

runtime: The runtime for this client, or the default if unset.

http_connect_proxy_config: Configuration for HTTP CONNECT proxy.

dns_load_balancing_config: DNS load balancing configuration for the

client connection. Default is to re-resolve DNS every 30s. Can

be set to ``None`` to disable. Silently disabled when

``http_connect_proxy_config`` is set, since the two are mutually

exclusive.

grpc_compression: Transport-level gRPC compression for the client

connection. Default is gzip. Set to

:py:attr:`GrpcCompression.NONE` to disable compression.

payload_limits: Warning thresholds for outbound payload/memo sizes. Over-threshold

fields are logged but still sent. Set a threshold to 0 to disable it.

header_codec_behavior: Encoding behavior for headers sent by the client.

"""

connect_config = temporalio.service.ConnectConfig(

target_host=target_host,

api_key=api_key,

tls=tls,

retry_config=retry_config,

keep_alive_config=keep_alive_config,

connect_timeout=connect_timeout,

rpc_metadata=rpc_metadata,

identity=identity or "",

lazy=lazy,

runtime=runtime,

http_connect_proxy_config=http_connect_proxy_config,

dns_load_balancing_config=dns_load_balancing_config,

grpc_compression=grpc_compression,

payload_limits=payload_limits,

)

def make_lambda(

plugin: Plugin, next: Callable[[ConnectConfig], Awaitable[ServiceClient]]

):

return lambda config: plugin.connect_service_client(config, next)

next_function = ServiceClient.connect

for plugin in reversed(plugins):

next_function = make_lambda(plugin, next_function)

service_client = await next_function(connect_config)

return cls(

service_client,

namespace=namespace,

data_converter=data_converter,

interceptors=interceptors,

default_workflow_query_reject_condition=default_workflow_query_reject_condition,

header_codec_behavior=header_codec_behavior,

plugins=plugins,

)

def __init__(

self,

service_client: temporalio.service.ServiceClient,

*,

namespace: str = "default",

data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default,

plugins: Sequence[Plugin] = [],

interceptors: Sequence[Interceptor] = [],

default_workflow_query_reject_condition: None

| (temporalio.common.QueryRejectCondition) = None,

header_codec_behavior: HeaderCodecBehavior = HeaderCodecBehavior.NO_CODEC,

):

"""Create a Temporal client from a service client.

See :py:meth:`connect` for details on the parameters.

"""

# Store the config for tracking

config = ClientConfig(

service_client=service_client,

namespace=namespace,

data_converter=data_converter,

plugins=plugins,

interceptors=interceptors,

default_workflow_query_reject_condition=default_workflow_query_reject_condition,

header_codec_behavior=header_codec_behavior,

)

self._initial_config = config.copy()

for plugin in plugins:

config = plugin.configure_client(config)

self._init_from_config(config)

def _init_from_config(self, config: ClientConfig):

self._config = config

# Iterate over interceptors in reverse building the impl

self._impl: OutboundInterceptor = _ClientImpl(self)

for interceptor in reversed(list(self._config["interceptors"])):

self._impl = interceptor.intercept_client(self._impl)

def config(self, *, active_config: bool = False) -> ClientConfig:

"""Config, as a dictionary, used to create this client.

Args:

active_config: If true, return the modified configuration in use rather than the initial one

provided to the client.

This makes a shallow copy of the config each call.

"""

config = self._config.copy() if active_config else self._initial_config.copy()

config["interceptors"] = list(config["interceptors"])

return config

@property

def service_client(self) -> temporalio.service.ServiceClient:

"""Raw gRPC service client."""

return self._config["service_client"]

@property

def workflow_service(self) -> temporalio.service.WorkflowService:

"""Raw gRPC workflow service client."""

return self._config["service_client"].workflow_service

@property

def operator_service(self) -> temporalio.service.OperatorService:

"""Raw gRPC operator service client."""

return self._config["service_client"].operator_service

@property

def test_service(self) -> temporalio.service.TestService:

"""Raw gRPC test service client."""

return self._config["service_client"].test_service

@property

def namespace(self) -> str:

"""Namespace used in calls by this client."""

return self._config["namespace"]

@property

def identity(self) -> str:

"""Identity used in calls by this client."""

return self._config["service_client"].config.identity

@property

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

"""Data converter used by this client."""

return self._config["data_converter"]

@property

def rpc_metadata(self) -> Mapping[str, str | bytes]:

"""Headers for every call made by this client.

Do not use mutate this mapping. Rather, set this property with an

entirely new mapping to change the headers.

"""

return self.service_client.config.rpc_metadata

@rpc_metadata.setter

def rpc_metadata(self, value: Mapping[str, str | bytes]) -> None:

"""Update the headers for this client.

Do not mutate this mapping after set. Rather, set an entirely new

mapping if changes are needed.

Raises:

TypeError: the key/value pair is not a valid gRPC ASCII or binary metadata.

All binary metadata must be supplied as bytes, and the key must end in '-bin'.

.. warning::

Attempting to set an invalid binary RPC metadata value may leave the client

in an inconsistent state (as well as raise a :py:class:`TypeError`).

"""

# Update config and perform update

# This may raise if the metadata is invalid:

self.service_client.update_rpc_metadata(value)

self.service_client.config.rpc_metadata = value

@property

def api_key(self) -> str | None:

"""API key for every call made by this client."""

return self.service_client.config.api_key

@api_key.setter

def api_key(self, value: str | None) -> None:

"""Update the API key for this client.

This is only set if RPCmetadata doesn't already have an "authorization"

key.

"""

# Update config and perform update

self.service_client.config.api_key = value

self.service_client.update_api_key(value)

# Overload for no-param workflow

@overload

async def start_workflow(

self,

workflow: MethodAsyncNoParam[SelfType, ReturnType],

*,

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> WorkflowHandle[SelfType, ReturnType]: ...

# Overload for single-param workflow

@overload

async def start_workflow(

self,

workflow: MethodAsyncSingleParam[SelfType, ParamType, ReturnType],

arg: ParamType,

*,

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> WorkflowHandle[SelfType, ReturnType]: ...

# Overload for multi-param workflow

@overload

async def start_workflow(

self,

workflow: Callable[

Concatenate[SelfType, MultiParamSpec], Awaitable[ReturnType]

],

*,

args: Sequence[Any],

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> WorkflowHandle[SelfType, ReturnType]: ...

# Overload for string-name workflow

@overload

async def start_workflow(

self,

workflow: str,

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: type | None = None,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> WorkflowHandle[Any, Any]: ...

async def start_workflow(

self,

workflow: str | Callable[..., Awaitable[Any]],

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: type | None = None,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

stack_level: int = 2,

) -> WorkflowHandle[Any, Any]:

"""Start a workflow and return its handle.

Args:

workflow: String name or class method decorated with

``@workflow.run`` for the workflow to start.

arg: Single argument to the workflow.

args: Multiple arguments to the workflow. Cannot be set if arg is.

id: Unique identifier for the workflow execution.

task_queue: Task queue to run the workflow on.

result_type: For string workflows, this can set the specific result

type hint to deserialize into.

execution_timeout: Total workflow execution timeout including

retries and continue as new.

run_timeout: Timeout of a single workflow run.

task_timeout: Timeout of a single workflow task.

id_conflict_policy: Behavior when a workflow is currently running with the same ID.

Default is UNSPECIFIED, which effectively means fail the start attempt.

Set to USE_EXISTING for idempotent deduplication on workflow ID.

With Temporal Server 1.32.0 or later, the returned handle is

scoped to the existing workflow's run chain.

Cannot be set if ``id_reuse_policy`` is set to TERMINATE_IF_RUNNING.

id_reuse_policy: Behavior when a closed workflow with the same ID exists.

Default is ALLOW_DUPLICATE.

retry_policy: Retry policy for the workflow.

cron_schedule: See https://docs.temporal.io/docs/content/what-is-a-temporal-cron-job/

memo: Memo for the workflow.

search_attributes: Search attributes for the workflow. The

dictionary form of this is deprecated, use

:py:class:`temporalio.common.TypedSearchAttributes`.

static_summary: A single-line fixed summary for this workflow execution that may appear

in the UI/CLI. This can be in single-line Temporal markdown format.

static_details: General fixed details for this workflow execution that may appear in

UI/CLI. This can be in Temporal markdown format and can span multiple lines. This is

a fixed value on the workflow that cannot be updated. For details that can be

updated, use :py:meth:`temporalio.workflow.get_current_details` within the workflow.

start_delay: Amount of time to wait before starting the workflow.

This does not work with ``cron_schedule``.

start_signal: If present, this signal is sent as signal-with-start

instead of traditional workflow start.

start_signal_args: Arguments for start_signal if start_signal

present.

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.

request_eager_start: Potentially reduce the latency to start this workflow by

encouraging the server to start it on a local worker running with

this same client.

priority: Priority of the workflow execution.

versioning_override: Overrides the versioning behavior for this workflow.

Returns:

A workflow handle to the started workflow.

Raises:

temporalio.exceptions.WorkflowAlreadyStartedError: Workflow has

already been started.

RPCError: Workflow could not be started for some other reason.

"""

temporalio.common._warn_on_deprecated_search_attributes(

search_attributes, stack_level=stack_level

)

name, result_type_from_type_hint = (

temporalio.workflow._Definition.get_name_and_result_type(workflow)

)

return await self._impl.start_workflow(

StartWorkflowInput(

workflow=name,

args=temporalio.common._arg_or_args(arg, args),

id=id,

task_queue=task_queue,

execution_timeout=execution_timeout,

run_timeout=run_timeout,

task_timeout=task_timeout,

id_reuse_policy=id_reuse_policy,

id_conflict_policy=id_conflict_policy,

retry_policy=retry_policy,

cron_schedule=cron_schedule,

memo=memo,

search_attributes=search_attributes,

start_delay=start_delay,

versioning_override=versioning_override,

headers={},

static_summary=static_summary,

static_details=static_details,

start_signal=start_signal,

start_signal_args=start_signal_args,

ret_type=result_type or result_type_from_type_hint,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

request_eager_start=request_eager_start,

priority=priority,

)

)

# Overload for no-param workflow

@overload

async def execute_workflow(

self,

workflow: MethodAsyncNoParam[SelfType, ReturnType],

*,

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> ReturnType: ...

# Overload for single-param workflow

@overload

async def execute_workflow(

self,

workflow: MethodAsyncSingleParam[SelfType, ParamType, ReturnType],

arg: ParamType,

*,

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> ReturnType: ...

# Overload for multi-param workflow

@overload

async def execute_workflow(

self,

workflow: Callable[

Concatenate[SelfType, MultiParamSpec], Awaitable[ReturnType]

],

*,

args: Sequence[Any],

id: str,

task_queue: str,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> ReturnType: ...

# Overload for string-name workflow

@overload

async def execute_workflow(

self,

workflow: str,

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: type | None = None,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> Any: ...

async def execute_workflow(

self,

workflow: str | Callable[..., Awaitable[Any]],

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: type | None = None,

execution_timeout: timedelta | None = None,

run_timeout: timedelta | None = None,

task_timeout: timedelta | None = None,

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,

id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy = temporalio.common.WorkflowIDConflictPolicy.UNSPECIFIED,

retry_policy: temporalio.common.RetryPolicy | None = None,

cron_schedule: str = "",

memo: Mapping[str, Any] | None = None,

search_attributes: None

| (

temporalio.common.TypedSearchAttributes | temporalio.common.SearchAttributes

) = None,

static_summary: str | None = None,

static_details: str | None = None,

start_delay: timedelta | None = None,

start_signal: str | None = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: timedelta | None = None,

request_eager_start: bool = False,

priority: temporalio.common.Priority = temporalio.common.Priority.default,

versioning_override: temporalio.common.VersioningOverride | None = None,

) -> Any:

"""Start a workflow and wait for completion.

This is a shortcut for :py:meth:`start_workflow` +

:py:meth:`WorkflowHandle.result`.

"""

return await (

# We have to tell MyPy to ignore errors here because we want to call

# the non-@overload form of this and MyPy does not support that

await self.start_workflow( # type: ignore

workflow, # type: ignore[arg-type]

arg,

args=args,

task_queue=task_queue,

result_type=result_type,

id=id,

execution_timeout=execution_timeout,

run_timeout=run_timeout,

task_timeout=task_timeout,

id_reuse_policy=id_reuse_policy,

id_conflict_policy=id_conflict_policy,

retry_policy=retry_policy,

cron_schedule=cron_schedule,

memo=memo,

search_attributes=search_attributes,

static_summary=static_summary,

static_details=static_details,

start_delay=start_delay,

start_signal=start_signal,

start_signal_args=start_signal_args,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

request_eager_start=request_eager_start,

priority=priority,

versioning_override=versioning_override,

stack_level=3,

)

).result()

def get_workflow_handle(

self,

workflow_id: str,

*,

run_id: str | None = None,

first_execution_run_id: str | None = None,

result_type: type | None = None,

) -> WorkflowHandle[Any, Any]:

"""Get a workflow handle to an existing workflow by its ID.

Args:

workflow_id: Workflow ID to get a handle to.

run_id: Run ID that will be used for all calls.

first_execution_run_id: First execution run ID used for cancellation

and termination.

result_type: The result type to deserialize into if known.

Returns:

The workflow handle.

"""

return WorkflowHandle(

self,

workflow_id,

run_id=run_id,

result_run_id=run_id,

first_execution_run_id=first_execution_run_id,

result_type=result_type,

)

def get_workflow_handle_for(

self,

workflow: (

MethodAsyncNoParam[SelfType, ReturnType]

| MethodAsyncSingleParam[SelfType, Any, ReturnType]

),

workflow_id: str,

*,

run_id: str | None = None,

first_execution_run_id: str | None = None,

) -> WorkflowHandle[SelfType, ReturnType]:

"""Get a typed workflow handle to an existing workflow by its ID.

This is the same as :py:meth:`get_workflow_handle` but typed.

Args:

workflow: The workflow run method to use for typing the handle.

workflow_id: Workflow ID to get a handle to.

run_id: Run ID that will be used for all calls.

first_execution_run_id: First execution run ID used for cancellation

and termination.

Returns:

The workflow handle.

"""

defn = temporalio.workflow._Definition.must_from_run_fn(workflow)

return self.get_workflow_handle(

workflow_id,

run_id=run_id,

first_execution_run_id=first_execution_run_id,

result_type=defn.ret_type,

)

# Overload for no-param update

@overload

async def execute_update_with_start_workflow(

self,

update: temporalio.workflow.UpdateMethodMultiParam[[SelfType], LocalReturnType],

*,

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

id: str | None = None,

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

rpc_timeout: timedelta | None = None,

) -> LocalReturnType: ...

# Overload for single-param update

@overload

async def execute_update_with_start_workflow(

self,

update: temporalio.workflow.UpdateMethodMultiParam[

[SelfType, ParamType], LocalReturnType

],

arg: ParamType,

*,

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

id: str | None = None,

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

rpc_timeout: timedelta | None = None,

) -> LocalReturnType: ...

# Overload for multi-param update

@overload

async def execute_update_with_start_workflow(

self,

update: temporalio.workflow.UpdateMethodMultiParam[

MultiParamSpec, LocalReturnType

],

*,

args: MultiParamSpec.args, # type: ignore

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

id: str | None = None,

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

rpc_timeout: timedelta | None = None,

) -> LocalReturnType: ...

# Overload for string-name update

@overload

async def execute_update_with_start_workflow(

self,

update: str,

arg: Any = temporalio.common._arg_unset,

*,

start_workflow_operation: WithStartWorkflowOperation[Any, Any],

args: Sequence[Any] = [],

id: str | None = None,

result_type: type | None = None,

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

rpc_timeout: timedelta | None = None,

) -> Any: ...

async def execute_update_with_start_workflow(

self,

update: str | Callable,

arg: Any = temporalio.common._arg_unset,

*,

start_workflow_operation: WithStartWorkflowOperation[Any, Any],

args: Sequence[Any] = [],

id: str | None = None,

result_type: type | None = None,

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

rpc_timeout: timedelta | None = None,

) -> Any:

"""Send an update-with-start request and wait for the update to complete.