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

Type,

Union,

cast,

overload,

)

import google.protobuf.duration_pb2

import google.protobuf.json_format

import google.protobuf.timestamp_pb2

from typing_extensions import Concatenate, 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.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.service import (

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,

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

"""

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,

)

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,

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,

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,

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

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,

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,

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

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,

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,

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

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,

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,

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

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,

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,

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

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

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.

This is currently experimental.

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.

"""

# Use definition if callable

name: str

if isinstance(workflow, str):

name = workflow

elif callable(workflow):

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

if not defn.name:

raise ValueError("Cannot invoke dynamic workflow explicitly")

name = defn.name

if result_type is None:

result_type = defn.ret_type

else:

raise TypeError("Workflow must be a string or callable")

temporalio.common._warn_on_deprecated_search_attributes(

search_attributes, stack_level=stack_level

)

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,

retry_policy=retry_policy,

cron_schedule=cron_schedule,

memo=memo,

search_attributes=search_attributes,

start_delay=start_delay,

headers={},

start_signal=start_signal,

start_signal_args=start_signal_args,

ret_type=result_type,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

request_eager_start=request_eager_start,

)

)

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

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,

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,

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

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,

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,

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

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,

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,

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

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,

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,

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

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,

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,

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

retry_policy=retry_policy,

cron_schedule=cron_schedule,

memo=memo,

search_attributes=search_attributes,

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,

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,

)

def list_workflows(

self,

query: Optional[str] = None,

*,

page_size: int = 1000,

next_page_token: Optional[bytes] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> WorkflowExecutionAsyncIterator:

"""List workflows.

This does not make a request until the first iteration is attempted.

Therefore any errors will not occur until then.

Args:

query: A Temporal visibility list filter. See Temporal documentation

concerning visibility list filters including behavior when left

unset.

page_size: Maximum number of results for each page.

next_page_token: A previously obtained next page token if doing

pagination. Usually not needed as the iterator automatically

starts from the beginning.

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

client-level RPC metadata keys.

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

Returns:

An async iterator that can be used with ``async for``.

"""

return self._impl.list_workflows(

ListWorkflowsInput(

query=query,

page_size=page_size,

next_page_token=next_page_token,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

)

)

@overload

def get_async_activity_handle(

self, *, workflow_id: str, run_id: Optional[str], activity_id: str

) -> AsyncActivityHandle:

pass

@overload

def get_async_activity_handle(self, *, task_token: bytes) -> AsyncActivityHandle:

pass

def get_async_activity_handle(

self,

*,

workflow_id: Optional[str] = None,

run_id: Optional[str] = None,

activity_id: Optional[str] = None,

task_token: Optional[bytes] = None,

) -> AsyncActivityHandle:

"""Get an async activity handle.

Either the workflow_id, run_id, and activity_id can be provided, or a

singular task_token can be provided.

Args:

workflow_id: Workflow ID for the activity. Cannot be set if

task_token is set.

run_id: Run ID for the activity. Cannot be set if task_token is set.

activity_id: ID for the activity. Cannot be set if task_token is

set.

task_token: Task token for the activity. Cannot be set if any of the

id parameters are set.

Returns:

A handle that can be used for completion or heartbeat.

"""

if task_token is not None:

if workflow_id is not None or run_id is not None or activity_id is not None:

raise ValueError("Task token cannot be present with other IDs")

return AsyncActivityHandle(self, task_token)

elif workflow_id is not None:

if activity_id is None:

raise ValueError(

"Workflow ID, run ID, and activity ID must all be given together"

)

return AsyncActivityHandle(

self,

AsyncActivityIDReference(

workflow_id=workflow_id, run_id=run_id, activity_id=activity_id

),

)

raise ValueError("Task token or workflow/run/activity ID must be present")

async def create_schedule(

self,

id: str,

schedule: Schedule,

*,

trigger_immediately: bool = False,

backfill: Sequence[ScheduleBackfill] = [],

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

search_attributes: Optional[

Union[

temporalio.common.TypedSearchAttributes,

temporalio.common.SearchAttributes,

]

] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> ScheduleHandle:

"""Create a schedule and return its handle.

Args:

id: Unique identifier of the schedule.

schedule: Schedule to create.

trigger_immediately: If true, trigger one action immediately when

creating the schedule.

backfill: Set of time periods to take actions on as if that time

passed right now.

memo: Memo for the schedule. Memo for a scheduled workflow is part

of the schedule action.

search_attributes: Search attributes for the schedule. Search

attributes for a scheduled workflow are part of the scheduled

action. The dictionary form of this is DEPRECATED, use

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

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.

Returns:

A handle to the created schedule.

Raises:

ScheduleAlreadyRunningError: If a schedule with this ID is already

running.

"""

temporalio.common._warn_on_deprecated_search_attributes(search_attributes)

return await self._impl.create_schedule(

CreateScheduleInput(

id=id,

schedule=schedule,

trigger_immediately=trigger_immediately,

backfill=backfill,

memo=memo,

search_attributes=search_attributes,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

)

)

def get_schedule_handle(self, id: str) -> ScheduleHandle:

"""Get a schedule handle for the given ID."""

return ScheduleHandle(self, id)

async def list_schedules(

self,

*,

page_size: int = 1000,

next_page_token: Optional[bytes] = None,

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

rpc_timeout: Optional[timedelta] = None,

) -> ScheduleAsyncIterator:

"""List schedules.

This does not make a request until the first iteration is attempted.

Therefore any errors will not occur until then.

Note, this list is eventually consistent. Therefore if a schedule is

added or deleted, it may not be available in the list immediately.

Args:

page_size: Maximum number of results for each page.

next_page_token: A previously obtained next page token if doing

pagination. Usually not needed as the iterator automatically

starts from the beginning.

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

client-level RPC metadata keys.

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

Returns:

An async iterator that can be used with ``async for``.

"""

return self._impl.list_schedules(

ListSchedulesInput(

page_size=page_size,

next_page_token=next_page_token,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

)

)

async def update_worker_build_id_compatibility(

self,

task_queue: str,

operation: BuildIdOp,

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

rpc_timeout: Optional[timedelta] = None,

) -> None:

"""Used to add new Build IDs or otherwise update the relative compatibility of Build Ids as

defined on a specific task queue for the Worker Versioning feature.

For more on this feature, see https://docs.temporal.io/workers#worker-versioning

.. warning::

This API is experimental

Args:

task_queue: The task queue to target.

operation: The operation to perform.

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

client-level RPC metadata keys.

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

"""

return await self._impl.update_worker_build_id_compatibility(

UpdateWorkerBuildIdCompatibilityInput(

task_queue,

operation,

rpc_metadata=rpc_metadata,

rpc_timeout=rpc_timeout,

)

)

async def get_worker_build_id_compatibility(

self,