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

from __future__ import annotations

import asyncio

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,

Iterator,

List,

Mapping,

MutableMapping,

NoReturn,

Optional,

Sequence,

Tuple,

Type,

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

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

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

...

@overload

def defn(

*, sandboxed: bool = True, dynamic: bool = False

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

...

def defn(

cls: Optional[ClassType] = None,

*,

name: Optional[str] = None,

sandboxed: bool = True,

dynamic: bool = False,

):

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

"""

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,

)

return cls

if cls is not None:

return decorator(cls)

return decorator

def run(fn: CallableAsyncType) -> CallableAsyncType:

"""Decorator for the workflow run method.

This must be set 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]

@overload

def signal(fn: CallableSyncOrAsyncReturnNoneType) -> CallableSyncOrAsyncReturnNoneType:

...

@overload

def signal(

*, name: str

) -> Callable[[CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType]:

...

@overload

def signal(

*, dynamic: Literal[True]

) -> Callable[[CallableSyncOrAsyncReturnNoneType], CallableSyncOrAsyncReturnNoneType]:

...

def signal(

fn: Optional[CallableSyncOrAsyncReturnNoneType] = None,

*,

name: Optional[str] = None,

dynamic: Optional[bool] = False,

):

"""Decorator for a workflow signal method.

This is set 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.

"""

def with_name(

name: Optional[str], fn: CallableSyncOrAsyncReturnNoneType

) -> CallableSyncOrAsyncReturnNoneType:

defn = _SignalDefinition(name=name, fn=fn, is_method=True)

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 name is not None or dynamic:

if name is not None and dynamic:

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

return partial(with_name, name)

if fn is None:

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

return with_name(fn.__name__, fn)

@overload

def query(fn: CallableType) -> CallableType:

...

@overload

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

...

@overload

def query(*, dynamic: Literal[True]) -> Callable[[CallableType], CallableType]:

...

def query(

fn: Optional[CallableType] = None,

*,

name: Optional[str] = None,

dynamic: Optional[bool] = False,

):

"""Decorator for a workflow query method.

This is set 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.

"""

def with_name(

name: Optional[str], fn: CallableType, *, bypass_async_check: bool = False

) -> CallableType:

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)

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:

if name is not None and dynamic:

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

return partial(with_name, name)

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 with_name(fn.__name__, fn, bypass_async_check=True)

@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]

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

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

"""

return _Runtime.current().workflow_get_current_build_id()

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

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]:

return getattr(asyncio.get_running_loop(), "__temporal_workflow_runtime", 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_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_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_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_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],

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

) -> 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_wait_condition(

self, fn: Callable[[], bool], *, timeout: Optional[float] = None

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

def metric_meter() -> temporalio.common.MetricMeter:

"""Get the metric meter for the current workflow.

This meter is replay safe which means that metrics will not be recorded

during replay.

Returns:

Current metric meter for this workflow for recording metrics.

"""

return _Runtime.current().workflow_metric_meter()

def now() -> datetime:

"""Current time from the workflow perspective.

This is the workflow equivalent of :py:func:`datetime.now` with the

:py:attr:`timezone.utc` parameter.

Returns:

UTC datetime for the current workflow time. The datetime does have UTC

set as the time zone.

"""

return datetime.fromtimestamp(time(), timezone.utc)

def patched(id: str) -> bool:

"""Patch a workflow.

When called, this will only return true if code should take the newer path

which means this is either not replaying or is replaying and has seen this

patch before.

Use :py:func:`deprecate_patch` when all workflows are done and will never be

queried again. The old code path can be used at that time too.

Args:

id: The identifier for this patch. This identifier may be used

repeatedly in the same workflow to represent the same patch

Returns:

True if this should take the newer path, false if it should take the

older path.

"""

return _Runtime.current().workflow_patch(id, deprecated=False)

def payload_converter() -> temporalio.converter.PayloadConverter:

"""Get the payload converter for the current workflow.

This is often used for dynamic workflows/signals/queries to convert

payloads.

"""

return _Runtime.current().workflow_payload_converter()

def random() -> Random:

"""Get a deterministic pseudo-random number generator.

Note, this random number generator is not cryptographically safe and should

not be used for security purposes.

Returns:

The deterministically-seeded pseudo-random number generator.

"""

return _Runtime.current().workflow_random()

def time() -> float:

"""Current seconds since the epoch from the workflow perspective.

This is the workflow equivalent of :py:func:`time.time`.

Returns:

Seconds since the epoch as a float.

"""

return time_ns() / 1e9

def time_ns() -> int:

"""Current nanoseconds since the epoch from the workflow perspective.

This is the workflow equivalent of :py:func:`time.time_ns`.

Returns:

Nanoseconds since the epoch

"""

return _Runtime.current().workflow_time_ns()

def upsert_search_attributes(

attributes: Union[

temporalio.common.SearchAttributes,

Sequence[temporalio.common.SearchAttributeUpdate],

]

) -> None:

"""Upsert search attributes for this workflow.

Args:

attributes: The attributes to set. This should be a sequence of

updates (i.e. values created via value_set and value_unset calls on

search attribute keys). The dictionary form of attributes is

DEPRECATED and if used, result in invalid key types on the

typed_search_attributes property in the info.

"""

temporalio.common._warn_on_deprecated_search_attributes(attributes)

_Runtime.current().workflow_upsert_search_attributes(attributes)

# Needs to be defined here to avoid a circular import

@runtime_checkable

class UpdateMethodMultiParam(Protocol[MultiParamSpec, ProtocolReturnType]):

"""Decorated workflow update functions implement this."""

_defn: temporalio.workflow._UpdateDefinition

def __call__(

self, *args: MultiParamSpec.args, **kwargs: MultiParamSpec.kwargs

) -> Union[ProtocolReturnType, Awaitable[ProtocolReturnType]]:

"""Generic callable type callback."""

...

def validator(

self, vfunc: Callable[MultiParamSpec, None]

) -> Callable[MultiParamSpec, None]:

"""Use to decorate a function to validate the arguments passed to the update handler."""

...

@overload

def update(

fn: Callable[MultiParamSpec, Awaitable[ReturnType]]

) -> UpdateMethodMultiParam[MultiParamSpec, ReturnType]:

...

@overload

def update(

fn: Callable[MultiParamSpec, ReturnType]

) -> UpdateMethodMultiParam[MultiParamSpec, ReturnType]:

...

@overload

def update(

*, name: str

) -> Callable[

[Callable[MultiParamSpec, ReturnType]],

UpdateMethodMultiParam[MultiParamSpec, ReturnType],

]:

...

@overload

def update(

*, dynamic: Literal[True]

) -> Callable[

[Callable[MultiParamSpec, ReturnType]],

UpdateMethodMultiParam[MultiParamSpec, ReturnType],

]:

...

def update(

fn: Optional[CallableSyncOrAsyncType] = None,

*,

name: Optional[str] = None,

dynamic: Optional[bool] = False,

):

"""Decorator for a workflow update handler method.

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

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

must be decorated.

You may also optionally define a validator method that will be called before

this handler you have applied this decorator to. You can specify the validator

with ``@update_handler_function_name.validator``.

Update methods can only have positional parameters. Best practice for

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

argument that can accept more fields later if needed. The handler may return

a serializable value which will be sent back to the caller of the update.

.. warning::

This API is experimental

Args:

fn: The function to decorate.

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

when ``dynamic`` is present.

dynamic: If true, this handles all updates 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.

"""

def with_name(

name: Optional[str], fn: CallableSyncOrAsyncType

) -> CallableSyncOrAsyncType:

defn = _UpdateDefinition(name=name, fn=fn, is_method=True)

if defn.dynamic_vararg:

raise RuntimeError(

"Dynamic updates do not support a vararg third param, use Sequence[RawValue]",

)

setattr(fn, "_defn", defn)

setattr(fn, "validator", partial(_update_validator, defn))

return fn

if name is not None or dynamic:

if name is not None and dynamic:

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

return partial(with_name, name)

if fn is None:

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

return with_name(fn.__name__, fn)

def _update_validator(

update_def: _UpdateDefinition, fn: Optional[Callable[..., None]] = None

) -> Optional[Callable[..., None]]:

"""Decorator for a workflow update validator method."""

if fn is not None:

update_def.set_validator(fn)

return fn

def uuid4() -> uuid.UUID:

"""Get a new, determinism-safe v4 UUID based on :py:func:`random`.

Note, this UUID is not cryptographically safe and should not be used for

security purposes.

Returns:

A deterministically-seeded v4 UUID.

"""

return uuid.UUID(bytes=random().getrandbits(16 * 8).to_bytes(16, "big"), version=4)

async def wait_condition(

fn: Callable[[], bool], *, timeout: Optional[Union[timedelta, float]] = None

) -> None:

"""Wait on a callback to become true.

This function returns when the callback returns true (invoked each loop

iteration) or the timeout has been reached.

Args:

fn: Non-async callback that accepts no parameters and returns a boolean.

timeout: Optional number of seconds to wait until throwing

:py:class:`asyncio.TimeoutError`.

"""

await _Runtime.current().workflow_wait_condition(

fn,

timeout=timeout.total_seconds() if isinstance(timeout, timedelta) else timeout,

)

_sandbox_unrestricted = threading.local()

_in_sandbox = threading.local()

_imports_passed_through = threading.local()

class unsafe:

"""Contains static methods that should not normally be called during

workflow execution except in advanced cases.

"""

def __init__(self) -> None: # noqa: D107

raise NotImplementedError