"""Utilities that can decorate or be called inside workflows."""

from __future__ import annotations

import asyncio

import contextvars

import dataclasses

import inspect

import logging

import threading

import uuid

import warnings

from abc import ABC, abstractmethod

from contextlib import contextmanager

from dataclasses import dataclass

from datetime import datetime, timedelta, timezone

from enum import Enum, IntEnum

from functools import partial

from random import Random

from typing import (

TYPE_CHECKING,

Any,

Awaitable,

Callable,

Dict,

Generic,

Iterable,

Iterator,

List,

Mapping,

MutableMapping,

NoReturn,

Optional,

Sequence,

Tuple,

Type,

TypeVar,

Union,

cast,

overload,

)

from typing_extensions import (

Concatenate,

Literal,

Protocol,

TypedDict,

runtime_checkable,

)

import temporalio.api.common.v1

import temporalio.bridge.proto.child_workflow

import temporalio.bridge.proto.workflow_commands

import temporalio.common

import temporalio.converter

import temporalio.exceptions

import temporalio.workflow

from .types import (

AnyType,

CallableAsyncNoParam,

CallableAsyncSingleParam,

CallableAsyncType,

CallableSyncNoParam,

CallableSyncOrAsyncReturnNoneType,

CallableSyncOrAsyncType,

CallableSyncSingleParam,

CallableType,

ClassType,

MethodAsyncNoParam,

MethodAsyncSingleParam,

MethodSyncNoParam,

MethodSyncOrAsyncNoParam,

MethodSyncOrAsyncSingleParam,

MethodSyncSingleParam,

MultiParamSpec,

ParamType,

ProtocolReturnType,

ReturnType,

SelfType,

)

@overload

def defn(cls: ClassType) -> ClassType: ...

@overload

def defn(

*,

name: Optional[str] = None,

sandboxed: bool = True,

failure_exception_types: Sequence[Type[BaseException]] = [],

versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,

) -> Callable[[ClassType], ClassType]: ...

@overload

def defn(

*,

sandboxed: bool = True,

dynamic: bool = False,

versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,

) -> Callable[[ClassType], ClassType]: ...

def defn(

cls: Optional[ClassType] = None,

*,

name: Optional[str] = None,

sandboxed: bool = True,

dynamic: bool = False,

failure_exception_types: Sequence[Type[BaseException]] = [],

versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,

):

"""Decorator for workflow classes.

This must be set on any registered workflow class (it is ignored if on a

base class).

Args:

cls: The class to decorate.

name: Name to use for the workflow. Defaults to class ``__name__``. This

cannot be set if dynamic is set.

sandboxed: Whether the workflow should run in a sandbox. Default is

true.

dynamic: If true, this activity will be dynamic. Dynamic workflows have

to accept a single 'Sequence[RawValue]' parameter. This cannot be

set to true if name is present.

failure_exception_types: The types of exceptions that, if a

workflow-thrown exception extends, will cause the workflow/update to

fail instead of suspending the workflow via task failure. These are

applied in addition to ones set on the worker constructor. If

``Exception`` is set, it effectively will fail a workflow/update in

all user exception cases. WARNING: This setting is experimental.

versioning_behavior: Specifies the versioning behavior to use for this workflow.

WARNING: This setting is experimental.

"""

def decorator(cls: ClassType) -> ClassType:

# This performs validation

_Definition._apply_to_class(

cls,

workflow_name=name or cls.__name__ if not dynamic else None,

sandboxed=sandboxed,

failure_exception_types=failure_exception_types,

versioning_behavior=versioning_behavior,

)

return cls

if cls is not None:

return decorator(cls)

return decorator

def init(

init_fn: CallableType,

) -> CallableType:

"""Decorator for the workflow init method.

This may be used on the __init__ method of the workflow class to specify

that it accepts the same workflow input arguments as the ``@workflow.run``

method. If used, the parameters of your __init__ and ``@workflow.run``

methods must be identical.

Args:

init_fn: The __init__ method to decorate.

"""

if init_fn.__name__ != "__init__":

raise ValueError("@workflow.init may only be used on the __init__ method")

setattr(init_fn, "__temporal_workflow_init", True)

return init_fn

def run(fn: CallableAsyncType) -> CallableAsyncType:

"""Decorator for the workflow run method.

This must be used on one and only one async method defined on the same class

as ``@workflow.defn``. This can be defined on a base class method but must

then be explicitly overridden and defined on the workflow class.

Run methods can only have positional parameters. Best practice is to only

take a single object/dataclass argument that can accept more fields later if

needed.

Args:

fn: The function to decorate.

"""

if not inspect.iscoroutinefunction(fn):

raise ValueError("Workflow run method must be an async function")

# Disallow local classes because we need to have the class globally

# referenceable by name

if "<locals>" in fn.__qualname__:

raise ValueError(

"Local classes unsupported, @workflow.run cannot be on a local class"

)

setattr(fn, "__temporal_workflow_run", True)

# TODO(cretz): Why is MyPy unhappy with this return?

return fn # type: ignore[return-value]

class HandlerUnfinishedPolicy(Enum):

"""Actions taken if a workflow terminates with running handlers.

Policy defining actions taken when a workflow exits while update or signal handlers are running.

The workflow exit may be due to successful return, failure, cancellation, or continue-as-new.

"""

WARN_AND_ABANDON = 1

"""Issue a warning in addition to abandoning."""

ABANDON = 2

"""Abandon the handler.

In the case of an update handler this means that the client will receive an error rather than

the update result."""

class UnfinishedUpdateHandlersWarning(RuntimeWarning):

"""The workflow exited before all update handlers had finished executing."""

class UnfinishedSignalHandlersWarning(RuntimeWarning):

"""The workflow exited before all signal handlers had finished executing."""

@overload

def signal(

fn: CallableSyncOrAsyncReturnNoneType,

) -> CallableSyncOrAsyncReturnNoneType: ...

@overload

def signal(

*,

unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,

description: Optional[str] = None,

) -> Callable[

[CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType

]: ...

@overload

def signal(

*,

name: str,

unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,

description: Optional[str] = None,

) -> Callable[

[CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType

]: ...

@overload

def signal(

*,

dynamic: Literal[True],

unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,

description: Optional[str] = None,

) -> Callable[

[CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType

]: ...

def signal(

fn: Optional[CallableSyncOrAsyncReturnNoneType] = None,

*,

name: Optional[str] = None,

dynamic: Optional[bool] = False,

unfinished_policy: HandlerUnfinishedPolicy = HandlerUnfinishedPolicy.WARN_AND_ABANDON,

description: Optional[str] = None,

):

"""Decorator for a workflow signal method.

This is used on any async or non-async method that you wish to be called upon

receiving a signal. If a function overrides one with this decorator, it too

must be decorated.

Signal methods can only have positional parameters. Best practice for

non-dynamic signal methods is to only take a single object/dataclass

argument that can accept more fields later if needed. Return values from

signal methods are ignored.

Args:

fn: The function to decorate.

name: Signal name. Defaults to method ``__name__``. Cannot be present

when ``dynamic`` is present.

dynamic: If true, this handles all signals not otherwise handled. The

parameters of the method must be self, a string name, and a

``*args`` positional varargs. Cannot be present when ``name`` is

present.

unfinished_policy: Actions taken if a workflow terminates with

a running instance of this handler.

description: A short description of the signal that may appear in the UI/CLI.

"""

def decorator(

name: Optional[str],

unfinished_policy: HandlerUnfinishedPolicy,

fn: CallableSyncOrAsyncReturnNoneType,

) -> CallableSyncOrAsyncReturnNoneType:

if not name and not dynamic:

name = fn.__name__

defn = _SignalDefinition(

name=name,

fn=fn,

is_method=True,

unfinished_policy=unfinished_policy,

description=description,

)

setattr(fn, "__temporal_signal_definition", defn)

if defn.dynamic_vararg:

warnings.warn(

"Dynamic signals with vararg third param is deprecated, use Sequence[RawValue]",

DeprecationWarning,

stacklevel=2,

)

return fn

if not fn:

if name is not None and dynamic:

raise RuntimeError("Cannot provide name and dynamic boolean")

return partial(decorator, name, unfinished_policy)

else:

return decorator(fn.__name__, unfinished_policy, fn)

@overload

def query(fn: CallableType) -> CallableType: ...

@overload

def query(

*, name: str, description: Optional[str] = None

) -> Callable[[CallableType], CallableType]: ...

@overload

def query(

*, dynamic: Literal[True], description: Optional[str] = None

) -> Callable[[CallableType], CallableType]: ...

@overload

def query(*, description: str) -> Callable[[CallableType], CallableType]: ...

def query(

fn: Optional[CallableType] = None,

*,

name: Optional[str] = None,

dynamic: Optional[bool] = False,

description: Optional[str] = None,

):

"""Decorator for a workflow query method.

This is used on any non-async method that expects to handle a query. If a

function overrides one with this decorator, it too must be decorated.

Query methods can only have positional parameters. Best practice for

non-dynamic query methods is to only take a single object/dataclass

argument that can accept more fields later if needed. The return value is

the resulting query value. Query methods must not mutate any workflow state.

Args:

fn: The function to decorate.

name: Query name. Defaults to method ``__name__``. Cannot be present

when ``dynamic`` is present.

dynamic: If true, this handles all queries not otherwise handled. The

parameters of the method should be self, a string name, and a

``Sequence[RawValue]``. An older form of this accepted vararg

parameters which will now warn. Cannot be present when ``name`` is

present.

description: A short description of the query that may appear in the UI/CLI.

"""

def decorator(

name: Optional[str],

description: Optional[str],

fn: CallableType,

*,

bypass_async_check: bool = False,

) -> CallableType:

if not name and not dynamic:

name = fn.__name__

if not bypass_async_check and inspect.iscoroutinefunction(fn):

warnings.warn(

"Queries as async def functions are deprecated",

DeprecationWarning,

stacklevel=2,

)

defn = _QueryDefinition(

name=name, fn=fn, is_method=True, description=description

)

setattr(fn, "__temporal_query_definition", defn)

if defn.dynamic_vararg:

warnings.warn(

"Dynamic queries with vararg third param is deprecated, use Sequence[RawValue]",

DeprecationWarning,

stacklevel=2,

)

return fn

if name is not None or dynamic or description:

if name is not None and dynamic:

raise RuntimeError("Cannot provide name and dynamic boolean")

return partial(decorator, name, description)

if fn is None:

raise RuntimeError("Cannot create query without function or name or dynamic")

if inspect.iscoroutinefunction(fn):

warnings.warn(

"Queries as async def functions are deprecated",

DeprecationWarning,

stacklevel=2,

)

return decorator(fn.__name__, description, fn, bypass_async_check=True)

@dataclass(frozen=True)

class DynamicWorkflowConfig:

"""Returned by functions using the :py:func:`dynamic_config` decorator, see it for more."""

failure_exception_types: Optional[Sequence[Type[BaseException]]] = None

"""The types of exceptions that, if a workflow-thrown exception extends, will cause the

workflow/update to fail instead of suspending the workflow via task failure. These are applied

in addition to ones set on the worker constructor. If ``Exception`` is set, it effectively will

fail a workflow/update in all user exception cases.

Always overrides the equivalent parameter on :py:func:`defn` if set not-None.

WARNING: This setting is experimental.

"""

versioning_behavior: temporalio.common.VersioningBehavior = (

temporalio.common.VersioningBehavior.UNSPECIFIED

)

"""Specifies the versioning behavior to use for this workflow.

Always overrides the equivalent parameter on :py:func:`defn`.

WARNING: This setting is experimental.

"""

def dynamic_config(

fn: MethodSyncNoParam[SelfType, DynamicWorkflowConfig],

) -> MethodSyncNoParam[SelfType, DynamicWorkflowConfig]:

"""Decorator to allow configuring a dynamic workflow's behavior.

Because dynamic workflows may conceptually represent more than one workflow type, it may be

desirable to have different settings for fields that would normally be passed to

:py:func:`defn`, but vary based on the workflow type name or other information available in

the workflow's context. This function will be called after the workflow's :py:func:`init`,

if it has one, but before the workflow's :py:func:`run` method.

The method must only take self as a parameter, and any values set in the class it returns will

override those provided to :py:func:`defn`.

Cannot be specified on non-dynamic workflows.

Args:

fn: The function to decorate.

"""

if inspect.iscoroutinefunction(fn):

raise ValueError("Workflow dynamic_config method must be synchronous")

params = list(inspect.signature(fn).parameters.values())

if len(params) != 1:

raise ValueError("Workflow dynamic_config method must only take self parameter")

# Add marker attribute

setattr(fn, "__temporal_workflow_dynamic_config", True)

return fn

@dataclass(frozen=True)

class Info:

"""Information about the running workflow.

Retrieved inside a workflow via :py:func:`info`. This object is immutable

with the exception of the :py:attr:`search_attributes` and

:py:attr:`typed_search_attributes` which is updated on

:py:func:`upsert_search_attributes`.

Note, required fields may be added here in future versions. This class

should never be constructed by users.

"""

attempt: int

continued_run_id: Optional[str]

cron_schedule: Optional[str]

execution_timeout: Optional[timedelta]

headers: Mapping[str, temporalio.api.common.v1.Payload]

namespace: str

parent: Optional[ParentInfo]

root: Optional[RootInfo]

priority: temporalio.common.Priority

"""The priority of this workflow execution. If not set, or this server predates priorities,

then returns a default instance."""

raw_memo: Mapping[str, temporalio.api.common.v1.Payload]

retry_policy: Optional[temporalio.common.RetryPolicy]

run_id: str

run_timeout: Optional[timedelta]

search_attributes: temporalio.common.SearchAttributes

"""Search attributes for the workflow.

.. deprecated::

Use :py:attr:`typed_search_attributes` instead.

"""

start_time: datetime

"""The start time of the first task executed by the workflow."""

task_queue: str

task_timeout: timedelta

typed_search_attributes: temporalio.common.TypedSearchAttributes

"""Search attributes for the workflow.

Note, this may have invalid values or be missing values if passing the

deprecated form of dictionary attributes to

:py:meth:`upsert_search_attributes`.

"""

workflow_id: str

workflow_start_time: datetime

"""The start time of the workflow based on the workflow initialization."""

workflow_type: str

def _logger_details(self) -> Mapping[str, Any]:

return {

# TODO(cretz): worker ID?

"attempt": self.attempt,

"namespace": self.namespace,

"run_id": self.run_id,

"task_queue": self.task_queue,

"workflow_id": self.workflow_id,

"workflow_type": self.workflow_type,

}

def get_current_build_id(self) -> str:

"""Get the Build ID of the worker which executed the current Workflow Task.

May be undefined if the task was completed by a worker without a Build ID. If this worker is

the one executing this task for the first time and has a Build ID set, then its ID will be

used. This value may change over the lifetime of the workflow run, but is deterministic and

safe to use for branching.

.. deprecated::

Use get_current_deployment_version instead.

"""

return _Runtime.current().workflow_get_current_build_id()

def get_current_deployment_version(

self,

) -> Optional[temporalio.common.WorkerDeploymentVersion]:

"""Get the deployment version of the worker which executed the current Workflow Task.

May be None if the task was completed by a worker without a deployment version or build

id. If this worker is the one executing this task for the first time and has a deployment

version set, then its ID will be used. This value may change over the lifetime of the

workflow run, but is deterministic and safe to use for branching.

"""

return _Runtime.current().workflow_get_current_deployment_version()

def get_current_history_length(self) -> int:

"""Get the current number of events in history.

Note, this value may not be up to date if accessed inside a query.

Returns:

Current number of events in history (up until the current task).

"""

return _Runtime.current().workflow_get_current_history_length()

def get_current_history_size(self) -> int:

"""Get the current byte size of history.

Note, this value may not be up to date if accessed inside a query.

Returns:

Current byte-size of history (up until the current task).

"""

return _Runtime.current().workflow_get_current_history_size()

def is_continue_as_new_suggested(self) -> bool:

"""Get whether or not continue as new is suggested.

Note, this value may not be up to date if accessed inside a query.

Returns:

True if the server is configured to suggest continue as new and it

is suggested.

"""

return _Runtime.current().workflow_is_continue_as_new_suggested()

@dataclass(frozen=True)

class ParentInfo:

"""Information about the parent workflow."""

namespace: str

run_id: str

workflow_id: str

@dataclass(frozen=True)

class RootInfo:

"""Information about the root workflow."""

run_id: str

workflow_id: str

@dataclass(frozen=True)

class UpdateInfo:

"""Information about a workflow update."""

id: str

"""Update ID."""

name: str

"""Update type name."""

@property

def _logger_details(self) -> Mapping[str, Any]:

"""Data to be included in string appended to default logging output."""

return {

"update_id": self.id,

"update_name": self.name,

}

class _Runtime(ABC):

@staticmethod

def current() -> _Runtime:

loop = _Runtime.maybe_current()

if not loop:

raise _NotInWorkflowEventLoopError("Not in workflow event loop")

return loop

@staticmethod

def maybe_current() -> Optional[_Runtime]:

try:

return getattr(

asyncio.get_running_loop(), "__temporal_workflow_runtime", None

)

except RuntimeError:

return None

@staticmethod

def set_on_loop(

loop: asyncio.AbstractEventLoop, runtime: Optional[_Runtime]

) -> None:

if runtime:

setattr(loop, "__temporal_workflow_runtime", runtime)

elif hasattr(loop, "__temporal_workflow_runtime"):

delattr(loop, "__temporal_workflow_runtime")

def __init__(self) -> None:

super().__init__()

self._logger_details: Optional[Mapping[str, Any]] = None

@property

def logger_details(self) -> Mapping[str, Any]:

if self._logger_details is None:

self._logger_details = self.workflow_info()._logger_details()

return self._logger_details

@abstractmethod

def workflow_all_handlers_finished(self) -> bool: ...

@abstractmethod

def workflow_continue_as_new(

self,

*args: Any,

workflow: Union[None, Callable, str],

task_queue: Optional[str],

run_timeout: Optional[timedelta],

task_timeout: Optional[timedelta],

retry_policy: Optional[temporalio.common.RetryPolicy],

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

search_attributes: Optional[

Union[

temporalio.common.SearchAttributes,

temporalio.common.TypedSearchAttributes,

]

],

versioning_intent: Optional[VersioningIntent],

) -> NoReturn: ...

@abstractmethod

def workflow_extern_functions(self) -> Mapping[str, Callable]: ...

@abstractmethod

def workflow_get_current_build_id(self) -> str: ...

@abstractmethod

def workflow_get_current_deployment_version(

self,

) -> Optional[temporalio.common.WorkerDeploymentVersion]: ...

@abstractmethod

def workflow_get_current_history_length(self) -> int: ...

@abstractmethod

def workflow_get_current_history_size(self) -> int: ...

@abstractmethod

def workflow_get_external_workflow_handle(

self, id: str, *, run_id: Optional[str]

) -> ExternalWorkflowHandle[Any]: ...

@abstractmethod

def workflow_get_query_handler(self, name: Optional[str]) -> Optional[Callable]: ...

@abstractmethod

def workflow_get_signal_handler(

self, name: Optional[str]

) -> Optional[Callable]: ...

@abstractmethod

def workflow_get_update_handler(

self, name: Optional[str]

) -> Optional[Callable]: ...

@abstractmethod

def workflow_get_update_validator(

self, name: Optional[str]

) -> Optional[Callable]: ...

@abstractmethod

def workflow_info(self) -> Info: ...

@abstractmethod

def workflow_instance(self) -> Any: ...

@abstractmethod

def workflow_is_continue_as_new_suggested(self) -> bool: ...

@abstractmethod

def workflow_is_replaying(self) -> bool: ...

@abstractmethod

def workflow_memo(self) -> Mapping[str, Any]: ...

@abstractmethod

def workflow_memo_value(

self, key: str, default: Any, *, type_hint: Optional[Type]

) -> Any: ...

@abstractmethod

def workflow_upsert_memo(self, updates: Mapping[str, Any]) -> None: ...

@abstractmethod

def workflow_metric_meter(self) -> temporalio.common.MetricMeter: ...

@abstractmethod

def workflow_patch(self, id: str, *, deprecated: bool) -> bool: ...

@abstractmethod

def workflow_payload_converter(self) -> temporalio.converter.PayloadConverter: ...

@abstractmethod

def workflow_random(self) -> Random: ...

@abstractmethod

def workflow_set_query_handler(

self, name: Optional[str], handler: Optional[Callable]

) -> None: ...

@abstractmethod

def workflow_set_signal_handler(

self, name: Optional[str], handler: Optional[Callable]

) -> None: ...

@abstractmethod

def workflow_set_update_handler(

self,

name: Optional[str],

handler: Optional[Callable],

validator: Optional[Callable],

) -> None: ...

@abstractmethod

def workflow_start_activity(

self,

activity: Any,

*args: Any,

task_queue: Optional[str],

result_type: Optional[Type],

schedule_to_close_timeout: Optional[timedelta],

schedule_to_start_timeout: Optional[timedelta],

start_to_close_timeout: Optional[timedelta],

heartbeat_timeout: Optional[timedelta],

retry_policy: Optional[temporalio.common.RetryPolicy],

cancellation_type: ActivityCancellationType,

activity_id: Optional[str],

versioning_intent: Optional[VersioningIntent],

summary: Optional[str] = None,

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

) -> ActivityHandle[Any]: ...

@abstractmethod

async def workflow_start_child_workflow(

self,

workflow: Any,

*args: Any,

id: str,

task_queue: Optional[str],

result_type: Optional[Type],

cancellation_type: ChildWorkflowCancellationType,

parent_close_policy: ParentClosePolicy,

execution_timeout: Optional[timedelta],

run_timeout: Optional[timedelta],

task_timeout: Optional[timedelta],

id_reuse_policy: temporalio.common.WorkflowIDReusePolicy,

retry_policy: Optional[temporalio.common.RetryPolicy],

cron_schedule: str,

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

search_attributes: Optional[

Union[

temporalio.common.SearchAttributes,

temporalio.common.TypedSearchAttributes,

]

],

versioning_intent: Optional[VersioningIntent],

static_summary: Optional[str] = None,

static_details: Optional[str] = None,

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

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

@abstractmethod

def workflow_start_local_activity(

self,

activity: Any,

*args: Any,

result_type: Optional[Type],

schedule_to_close_timeout: Optional[timedelta],

schedule_to_start_timeout: Optional[timedelta],

start_to_close_timeout: Optional[timedelta],

retry_policy: Optional[temporalio.common.RetryPolicy],

local_retry_threshold: Optional[timedelta],

cancellation_type: ActivityCancellationType,

activity_id: Optional[str],

) -> ActivityHandle[Any]: ...

@abstractmethod

def workflow_time_ns(self) -> int: ...

@abstractmethod

def workflow_upsert_search_attributes(

self,

attributes: Union[

temporalio.common.SearchAttributes,

Sequence[temporalio.common.SearchAttributeUpdate],

],

) -> None: ...

@abstractmethod

async def workflow_sleep(

self, duration: float, *, summary: Optional[str] = None

) -> None: ...

@abstractmethod

async def workflow_wait_condition(

self,

fn: Callable[[], bool],

*,

timeout: Optional[float] = None,

timeout_summary: Optional[str] = None,

) -> None: ...

@abstractmethod

def workflow_get_current_details(self) -> str: ...

@abstractmethod

def workflow_set_current_details(self, details: str): ...

_current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(

"__temporal_current_update_info"

)

def _set_current_update_info(info: UpdateInfo) -> None:

_current_update_info.set(info)

def current_update_info() -> Optional[UpdateInfo]:

"""Info for the current update if any.

This is powered by :py:mod:`contextvars` so it is only valid within the

update handler and coroutines/tasks it has started.

Returns:

Info for the current update handler the code calling this is executing

within if any.

"""

return _current_update_info.get(None)

def deprecate_patch(id: str) -> None:

"""Mark a patch as deprecated.

This marks a workflow that had :py:func:`patched` in a previous version of

the code as no longer applicable because all workflows that use the old code

path are done and will never be queried again. Therefore the old code path

is removed as well.

Args:

id: The identifier originally used with :py:func:`patched`.

"""

_Runtime.current().workflow_patch(id, deprecated=True)

def extern_functions() -> Mapping[str, Callable]:

"""External functions available in the workflow sandbox.

Returns:

Mapping of external functions that can be called from inside a workflow

sandbox.

"""

return _Runtime.current().workflow_extern_functions()

def info() -> Info:

"""Current workflow's info.

Returns:

Info for the currently running workflow.

"""

return _Runtime.current().workflow_info()

def instance() -> Any:

"""Current workflow's instance.

Returns:

The currently running workflow instance.

"""

return _Runtime.current().workflow_instance()

def in_workflow() -> bool:

"""Whether the code is currently running in a workflow."""

return _Runtime.maybe_current() is not None

def memo() -> Mapping[str, Any]:

"""Current workflow's memo values, converted without type hints.

Since type hints are not used, the default converted values will come back.

For example, if the memo was originally created with a dataclass, the value

will be a dict. To convert using proper type hints, use

:py:func:`memo_value`.

Returns:

Mapping of all memo keys and they values without type hints.

"""

return _Runtime.current().workflow_memo()

@overload

def memo_value(key: str, default: Any = temporalio.common._arg_unset) -> Any: ...

@overload

def memo_value(key: str, *, type_hint: Type[ParamType]) -> ParamType: ...

@overload

def memo_value(

key: str, default: AnyType, *, type_hint: Type[ParamType]

) -> Union[AnyType, ParamType]: ...

def memo_value(

key: str,

default: Any = temporalio.common._arg_unset,

*,

type_hint: Optional[Type] = None,

) -> Any:

"""Memo value for the given key, optional default, and optional type

hint.

Args:

key: Key to get memo value for.

default: Default to use if key is not present. If unset, a

:py:class:`KeyError` is raised when the key does not exist.

type_hint: Type hint to use when converting.

Returns:

Memo value, converted with the type hint if present.

Raises:

KeyError: Key not present and default not set.

"""

return _Runtime.current().workflow_memo_value(key, default, type_hint=type_hint)