temporalio.workflow._context

module documentation
(source)

Undocumented

Class Info Information about the running workflow.
Class ParentInfo Information about the parent workflow.
Class RootInfo Information about the root workflow.
Class UpdateInfo Information about a workflow update.
Function current_update_info Info for the current update if any.
Function deprecate_patch Mark a patch as deprecated.
Function extern_functions External functions available in the workflow sandbox.
Function get_current_details Get the current details of the workflow which may appear in the UI/CLI. Unlike static details set at start, this value can be updated throughout the life of the workflow and is independent of the static details...
Function get_last_completion_result Get the result of the last run of the workflow. This will be None if there was no previous completion or the result was None. has_last_completion_result() can be used to differentiate.
Function get_last_failure Get the last failure of the workflow if it has run previously.
Function has_last_completion_result Gets whether there is a last completion result of the workflow.
Function in_workflow Whether the code is currently running in a workflow.
Function info Current workflow's info.
Function instance Current workflow's instance.
Function is_failure_exception Checks if the given exception is a workflow failure in the current workflow.
Function memo Current workflow's memo values, converted without type hints.
Function memo_value Memo value for the given key, optional default, and optional type hint.
Function metric_meter Get the metric meter for the current workflow.
Function new_random Create a Random instance that automatically reseeds when the workflow seed changes.
Function now Current time from the workflow perspective.
Function patched Patch a workflow.
Function payload_converter Get the payload converter for the current workflow.
Function random Get a deterministic pseudo-random number generator.
Function random_seed Get the current random seed value from core.
Function register_random_seed_callback Register a callback to be notified when the random seed changes.
Function set_current_details Set the current details of the workflow which may appear in the UI/CLI. Unlike static details set at start, this value can be updated throughout the life of the workflow and is independent of the static details...
Async Function sleep Sleep for the given duration.
Function time Current seconds since the epoch from the workflow perspective.
Function time_ns Current nanoseconds since the epoch from the workflow perspective.
Function upsert_memo Adds, modifies, and/or removes memos, with upsert semantics.
Function upsert_search_attributes Upsert search attributes for this workflow.
Function uuid4 Get a new, determinism-safe v4 UUID based on random.
Async Function wait_condition Wait on a callback to become true.
def current_update_info() -> UpdateInfo | None: (source)

Info for the current update if any.

This is powered by contextvars so it is only valid within the update handler and coroutines/tasks it has started.

Returns
UpdateInfo | None
Info for the current update handler the code calling this is executing
within if any.
def deprecate_patch(id: str): (source)

Mark a patch as deprecated.

This marks a workflow that had 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.

Parameters
id:strThe identifier originally used with patched.
def extern_functions() -> Mapping[str, Callable]: (source)

External functions available in the workflow sandbox.

Returns
Mapping[str, Callable]Mapping of external functions that can be called from inside a workflow sandbox.
def get_current_details() -> str: (source)

Get the current details of the workflow which may appear in the UI/CLI. Unlike static details set at start, this value can be updated throughout the life of the workflow and is independent of the static details. This can be in Temporal markdown format and can span multiple lines.

def get_last_completion_result() -> Any | None:
def get_last_completion_result(type_hint: type[ParamType]) -> ParamType | None:
(source)

Get the result of the last run of the workflow. This will be None if there was no previous completion or the result was None. has_last_completion_result() can be used to differentiate.

def get_last_failure() -> BaseException | None: (source)

Get the last failure of the workflow if it has run previously.

def has_last_completion_result() -> bool: (source)

Gets whether there is a last completion result of the workflow.

def in_workflow() -> bool: (source)

Whether the code is currently running in a workflow.

def info() -> Info: (source)

Current workflow's info.

Returns
InfoInfo for the currently running workflow.
def instance() -> Any: (source)

Current workflow's instance.

Returns
AnyThe currently running workflow instance.
def is_failure_exception(err: BaseException) -> bool: (source)

Checks if the given exception is a workflow failure in the current workflow.

Returns
boolTrue if the given exception is a workflow failure in the current workflow.
def memo() -> Mapping[str, Any]: (source)

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

Returns
Mapping[str, Any]Mapping of all memo keys and they values without type hints.
def memo_value(key: str, default: Any = temporalio.common._arg_unset) -> Any:
def memo_value(key: str, *, type_hint: type[ParamType]) -> ParamType:
def memo_value(key: str, default: AnyType, *, type_hint: type[ParamType]) -> AnyType | ParamType:
(source)

Memo value for the given key, optional default, and optional type hint.

Parameters
key:strKey to get memo value for.
default:AnyDefault to use if key is not present. If unset, a KeyError is raised when the key does not exist.
type_hint:type | NoneType hint to use when converting.
Returns
AnyMemo value, converted with the type hint if present.
Raises
KeyErrorKey not present and default not set.
def metric_meter() -> temporalio.common.MetricMeter: (source)

Get the metric meter for the current workflow.

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

Returns
temporalio.common.MetricMeterCurrent metric meter for this workflow for recording metrics.
def new_random() -> Random: (source)

Create a Random instance that automatically reseeds when the workflow seed changes.

This creates a new Random instance that is initially seeded with the current workflow seed, and automatically registers a callback to reseed itself whenever the workflow receives a new seed from core.

Returns
RandomA Random instance that stays synchronized with the workflow's randomness.
def now() -> datetime: (source)

Current time from the workflow perspective.

This is the workflow equivalent of datetime.now with the timezone.utc parameter.

Returns
datetimeUTC datetime for the current workflow time. The datetime does have UTC set as the time zone.
def patched(id: str) -> bool: (source)

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 deprecate_patch when all workflows are done and will never be queried again. The old code path can be used at that time too.

Parameters
id:strThe identifier for this patch. This identifier may be used repeatedly in the same workflow to represent the same patch
Returns
boolTrue if this should take the newer path, false if it should take the older path.
def payload_converter() -> temporalio.converter.PayloadConverter: (source)

Get the payload converter for the current workflow.

The returned converter has temporalio.converter.WorkflowSerializationContext set. This is often used for dynamic workflows/signals/queries to convert payloads.

def random() -> Random: (source)

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
RandomThe deterministically-seeded pseudo-random number generator.
def random_seed() -> int: (source)

Get the current random seed value from core.

This returns the seed value currently being used by the workflow's deterministic random number generator.

Returns
intThe current random seed as an integer.
def register_random_seed_callback(callback: Callable[[int], None]): (source)

Register a callback to be notified when the random seed changes.

The callback will be invoked whenever the workflow receives a new random seed from the core. This is useful for maintaining external random number generators that need to stay in sync with the workflow's randomness.

Parameters
callback:Callable[[int], None]Function to be called with the new seed value when it changes.
def set_current_details(description: str): (source)

Set the current details of the workflow which may appear in the UI/CLI. Unlike static details set at start, this value can be updated throughout the life of the workflow and is independent of the static details. This can be in Temporal markdown format and can span multiple lines.

async def sleep(duration: float | timedelta, *, summary: str | None = None): (source)

Sleep for the given duration.

Parameters
duration:float | timedeltaDuration to sleep in seconds or as a timedelta.
summary:str | NoneA single-line fixed summary for this timer that may appear in UI/CLI. This can be in single-line Temporal markdown format.
def time() -> float: (source)

Current seconds since the epoch from the workflow perspective.

This is the workflow equivalent of time.time.

Returns
floatSeconds since the epoch as a float.
def time_ns() -> int: (source)

Current nanoseconds since the epoch from the workflow perspective.

This is the workflow equivalent of time.time_ns.

Returns
intNanoseconds since the epoch
def upsert_memo(updates: Mapping[str, Any]): (source)

Adds, modifies, and/or removes memos, with upsert semantics.

Every memo that has a matching key has its value replaced with the one specified in updates. If the value is set to None, the memo is removed instead. For every key with no existing memo, a new memo is added with specified value (unless the value is None). Memos with keys not included in updates remain unchanged.

def upsert_search_attributes(attributes: temporalio.common.SearchAttributes | Sequence[temporalio.common.SearchAttributeUpdate]): (source)

Upsert search attributes for this workflow.

Parameters
attributes:temporalio.common.SearchAttributes | Sequence[temporalio.common.SearchAttributeUpdate]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.
def uuid4() -> uuid.UUID: (source)

Get a new, determinism-safe v4 UUID based on random.

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

Returns
uuid.UUIDA deterministically-seeded v4 UUID.
async def wait_condition(fn: Callable[[], bool], *, timeout: timedelta | float | None = None, timeout_summary: str | None = None): (source)

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.

Parameters
fn:Callable[[], bool]Non-async callback that accepts no parameters and returns a boolean.
timeout:timedelta | float | NoneOptional number of seconds to wait until throwing asyncio.TimeoutError.
timeout_summary:str | NoneOptional simple string identifying the timer (created if timeout is present) that may be visible in UI/CLI. While it can be normal text, it is best to treat as a timer ID.