"""Client for accessing Temporal."""

from __future__ import annotations

import asyncio

import copy

import dataclasses

import inspect

import json

import re

import uuid

import warnings

from abc import ABC, abstractmethod

from asyncio import Future

from dataclasses import dataclass

from datetime import datetime, timedelta, timezone

from enum import Enum, IntEnum

from typing import (

Any,

AsyncIterator,

Awaitable,

Callable,

Dict,

FrozenSet,

Generic,

Iterable,

Mapping,

Optional,

Sequence,

Tuple,

Type,

Union,

cast,

overload,

)

import google.protobuf.duration_pb2

import google.protobuf.json_format

import google.protobuf.timestamp_pb2

from typing_extensions import Concatenate, Required, TypedDict

import temporalio.api.common.v1

import temporalio.api.enums.v1

import temporalio.api.errordetails.v1

import temporalio.api.failure.v1

import temporalio.api.history.v1

import temporalio.api.schedule.v1

import temporalio.api.sdk.v1

import temporalio.api.taskqueue.v1

import temporalio.api.update.v1

import temporalio.api.workflow.v1

import temporalio.api.workflowservice.v1

import temporalio.common

import temporalio.converter

import temporalio.exceptions

import temporalio.runtime

import temporalio.service

import temporalio.workflow

from temporalio.activity import ActivityCancellationDetails

from temporalio.service import (

HttpConnectProxyConfig,

KeepAliveConfig,

RetryConfig,

RPCError,

RPCStatusCode,

TLSConfig,

)

from .types import (

AnyType,

LocalReturnType,

MethodAsyncNoParam,

MethodAsyncSingleParam,

MethodSyncOrAsyncNoParam,

MethodSyncOrAsyncSingleParam,

MultiParamSpec,

ParamType,

ReturnType,

SelfType,

)

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.

"""

@staticmethod

async def connect(

target_host: str,

*,

namespace: str = "default",

api_key: Optional[str] = None,

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

interceptors: Sequence[Interceptor] = [],

default_workflow_query_reject_condition: Optional[

temporalio.common.QueryRejectCondition

] = None,

tls: Union[bool, TLSConfig] = False,

retry_config: Optional[RetryConfig] = None,

keep_alive_config: Optional[KeepAliveConfig] = KeepAliveConfig.default,

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

identity: Optional[str] = None,

lazy: bool = False,

runtime: Optional[temporalio.runtime.Runtime] = None,

http_connect_proxy_config: Optional[HttpConnectProxyConfig] = None,

) -> Client:

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

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 false, the default, 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.

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.

"""

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,

rpc_metadata=rpc_metadata,

identity=identity or "",

lazy=lazy,

runtime=runtime,

http_connect_proxy_config=http_connect_proxy_config,

)

return Client(

await temporalio.service.ServiceClient.connect(connect_config),

namespace=namespace,

data_converter=data_converter,

interceptors=interceptors,

default_workflow_query_reject_condition=default_workflow_query_reject_condition,

)

def __init__(

self,

service_client: temporalio.service.ServiceClient,

*,

namespace: str = "default",

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

interceptors: Sequence[Interceptor] = [],

default_workflow_query_reject_condition: Optional[

temporalio.common.QueryRejectCondition

] = None,

):

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

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

"""

# Iterate over interceptors in reverse building the impl

self._impl: OutboundInterceptor = _ClientImpl(self)

for interceptor in reversed(list(interceptors)):

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

# Store the config for tracking

self._config = ClientConfig(

service_client=service_client,

namespace=namespace,

data_converter=data_converter,

interceptors=interceptors,

default_workflow_query_reject_condition=default_workflow_query_reject_condition,

)

def config(self) -> ClientConfig:

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

This makes a shallow copy of the config each call.

"""

config = self._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]:

"""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]) -> None:

"""Update the headers for this client.

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

mapping if changes are needed.

"""

# Update config and perform update

self.service_client.config.rpc_metadata = value

self.service_client.update_rpc_metadata(value)

@property

def api_key(self) -> Optional[str]:

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

return self.service_client.config.api_key

@api_key.setter

def api_key(self, value: Optional[str]) -> 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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[Type] = None,

execution_timeout: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = None,

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

async def start_workflow(

self,

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

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: Optional[Type] = None,

execution_timeout: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

stack_level: int = 2,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = None,

) -> 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_reuse_policy: How already-existing IDs are treated.

id_conflict_policy: How already-running workflows of the same ID are

treated. Default is unspecified which effectively means fail the

start attempt. This cannot be set if ``id_reuse_policy`` is set

to terminate if running.

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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[Type] = None,

execution_timeout: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = None,

) -> Any: ...

async def execute_workflow(

self,

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

arg: Any = temporalio.common._arg_unset,

*,

args: Sequence[Any] = [],

id: str,

task_queue: str,

result_type: Optional[Type] = None,

execution_timeout: Optional[timedelta] = None,

run_timeout: Optional[timedelta] = None,

task_timeout: Optional[timedelta] = None,

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

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

retry_policy: Optional[temporalio.common.RetryPolicy] = None,

cron_schedule: str = "",

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

start_delay: Optional[timedelta] = None,

start_signal: Optional[str] = None,

start_signal_args: Sequence[Any] = [],

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

rpc_timeout: Optional[timedelta] = None,

request_eager_start: bool = False,

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

versioning_override: Optional[temporalio.common.VersioningOverride] = 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: Optional[str] = None,

first_execution_run_id: Optional[str] = None,

result_type: Optional[Type] = 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: Union[

MethodAsyncNoParam[SelfType, ReturnType],

MethodAsyncSingleParam[SelfType, Any, ReturnType],

],

workflow_id: str,

*,

run_id: Optional[str] = None,

first_execution_run_id: Optional[str] = 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: Optional[str] = None,

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

rpc_timeout: Optional[timedelta] = 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: Optional[str] = None,

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

rpc_timeout: Optional[timedelta] = 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, # pyright: ignore

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

id: Optional[str] = None,

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

rpc_timeout: Optional[timedelta] = 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: Optional[str] = None,

result_type: Optional[Type] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> Any: ...

async def execute_update_with_start_workflow(

self,

update: Union[str, Callable],

arg: Any = temporalio.common._arg_unset,

*,

start_workflow_operation: WithStartWorkflowOperation[Any, Any],

args: Sequence[Any] = [],

id: Optional[str] = None,

result_type: Optional[Type] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> Any:

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

A WorkflowIDConflictPolicy must be set in the start_workflow_operation. If the

specified workflow execution is not running, a new workflow execution is started

and the update is sent in the first workflow task. Alternatively if the specified

workflow execution is running then, if the WorkflowIDConflictPolicy is

USE_EXISTING, the update is issued against the specified workflow, and if the

WorkflowIDConflictPolicy is FAIL, an error is returned. This call will block until

the update has completed, and return the update result. Note that this means that

the call will not return successfully until the update has been delivered to a

worker.

.. warning::

This API is experimental

Args:

update: Update function or name on the workflow. arg: Single argument to the

update.

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

start_workflow_operation: a WithStartWorkflowOperation definining the

WorkflowIDConflictPolicy and how to start the workflow in the event that a

workflow is started.

id: ID of the update. If not set, the default is a new UUID.

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

type hint to deserialize into.

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.

Raises:

WorkflowUpdateFailedError: If the update failed.

WorkflowUpdateRPCTimeoutOrCancelledError: This update call timed out

or was cancelled. This doesn't mean the update itself was timed out or

cancelled.

RPCError: There was some issue starting the workflow or sending the update to

the workflow.

"""

handle = await self._start_update_with_start(

update,

arg,

args=args,

start_workflow_operation=start_workflow_operation,

wait_for_stage=WorkflowUpdateStage.COMPLETED,

id=id,

result_type=result_type,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

)

return await handle.result()

# Overload for no-param start update

@overload

async def start_update_with_start_workflow(

self,

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

*,

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

wait_for_stage: WorkflowUpdateStage,

id: Optional[str] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> WorkflowUpdateHandle[LocalReturnType]: ...

# Overload for single-param start update

@overload

async def start_update_with_start_workflow(

self,

update: temporalio.workflow.UpdateMethodMultiParam[

[SelfType, ParamType], LocalReturnType

],

arg: ParamType,

*,

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

wait_for_stage: WorkflowUpdateStage,

id: Optional[str] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> WorkflowUpdateHandle[LocalReturnType]: ...

# Overload for multi-param start update

@overload

async def start_update_with_start_workflow(

self,

update: temporalio.workflow.UpdateMethodMultiParam[

MultiParamSpec, LocalReturnType

],

*,

args: MultiParamSpec.args, # pyright: ignore

start_workflow_operation: WithStartWorkflowOperation[SelfType, Any],

wait_for_stage: WorkflowUpdateStage,

id: Optional[str] = None,