Workflow¶

Workflow retry and timeout settings are durable command budgets. Use ActivityRetryPolicy on ctx.schedule_activity(...) for activity attempts and ChildWorkflowRetryPolicy on ctx.start_child_workflow(...) for child workflow attempts. Use TransportRetryPolicy only for client HTTP retries.

Use ctx.call_nexus_service(...) from workflow code to start a Nexus service operation durably. The Python worker records the accepted response or typed service failure as a side-effect marker, then replay resumes with NexusOperationResult or raises NexusOperationFailed at the yield point.

Yield a list for deterministic parallel composition. Lists may nest and may mix activity, child-workflow, and timer commands; the resolved value has the same nested shape and input order. Every leaf emits the shared parallel_group_* metadata and uses its ordinary Server command.

Use ctx.saga().run(forward) for sequential reverse-order compensation. Register each compensation only after its forward activity completes. The helper compensates on failure or cooperative cancellation and raises SagaCompensationFailed if compensation itself fails.

workflow ¶

Workflow authoring primitives: decorators, context, commands, and replayer.

A workflow is a Python class registered with defn. Its run method is a generator that yields command dataclasses (ScheduleActivity, StartTimer, StartChildWorkflow, ) the worker's replayer drives the generator forward by resolving each yielded command against the current history of the workflow run. Yield a list of commands to run them in parallel.

Determinism-sensitive helpers live on the WorkflowContext passed to run: WorkflowContext.random, WorkflowContext.uuid4, WorkflowContext.uuid7, WorkflowContext.now, WorkflowContext.patched, WorkflowContext.deprecate_patch, and WorkflowContext.side_effect all produce values that are recorded on first execution and replayed verbatim on every subsequent replay of the same history.

ActivityRetryPolicy `dataclass` ¶

ActivityRetryPolicy(max_attempts=3, initial_interval_seconds=1.0, backoff_coefficient=2.0, maximum_interval_seconds=None, non_retryable_error_types=list(), backoff_seconds=None)

Retry policy applied to one scheduled activity call.

The policy is snapped onto the durable activity execution when the workflow task completes, so later code deploys do not change the retry budget for an already-scheduled activity. It is a server-side durable retry policy, not the SDK HTTP transport retry policy.

non_retryable_error_types names failure types that should bypass this retry budget. An activity worker can also report non_retryable=True on a failure to stop retrying that activity execution.

ChildWorkflowRetryPolicy `dataclass` ¶

ChildWorkflowRetryPolicy(max_attempts=3, initial_interval_seconds=1.0, backoff_coefficient=2.0, maximum_interval_seconds=None, non_retryable_error_types=list(), backoff_seconds=None)

Bases: ActivityRetryPolicy

Retry policy applied to one started child workflow call.

This is recorded with the child workflow command and controls durable server-side child attempts. It is separate from SDK HTTP transport retry and from activity retry.

ScheduleActivity `dataclass` ¶

ScheduleActivity(activity_type, arguments, queue=None, retry_policy=None, start_to_close_timeout=None, schedule_to_start_timeout=None, schedule_to_close_timeout=None, heartbeat_timeout=None)

Command requesting an activity task.

Timeout fields are activity budgets, not HTTP request timeouts: start_to_close_timeout limits one activity attempt, schedule_to_start_timeout limits queue wait before an attempt starts, schedule_to_close_timeout limits the whole activity execution including retries, and heartbeat_timeout limits the gap between activity heartbeats.

StartTimer `dataclass` ¶

StartTimer(delay_seconds)

Command requesting a durable timer.

CompleteWorkflow `dataclass` ¶

CompleteWorkflow(result)

Command completing a workflow with a payload result.

FailWorkflow `dataclass` ¶

FailWorkflow(message, exception_type=None, exception_class=None, exception=None, non_retryable=False)

Command failing a workflow with diagnostic metadata.

CompleteUpdate `dataclass` ¶

CompleteUpdate(update_id, result)

Worker command completing an accepted workflow update.

FailUpdate `dataclass` ¶

FailUpdate(update_id, message, exception_type=None, exception_class=None, non_retryable=True)

Worker command failing an accepted workflow update.

ContinueAsNew `dataclass` ¶

ContinueAsNew(workflow_type=None, arguments=list(), task_queue=None)

Workflow return value that starts a new run with fresh history.

RecordSideEffect `dataclass` ¶

RecordSideEffect(result, workflow_stream=None)

Command recording the result of a non-deterministic function.

StartChildWorkflow `dataclass` ¶

StartChildWorkflow(workflow_type, arguments=list(), task_queue=None, parent_close_policy=None, retry_policy=None, execution_timeout_seconds=None, run_timeout_seconds=None)

Command requesting a child workflow run.

execution_timeout_seconds limits the overall child workflow execution. run_timeout_seconds limits one child run. These budgets are durable server-side workflow budgets and are separate from client HTTP timeouts.

NexusServiceCall `dataclass` ¶

NexusServiceCall(endpoint_name, service_name, operation_name, arguments=list(), idempotency_key=None, payload_codec=None, mode='sync', wait_for='completed', wait_timeout_seconds=None, caller_namespace=None, service_sdk_language=None, artifact_tuple=None, published_artifact_worker_execution=None, target_workflow_instance_id=None, target_workflow_run_id=None, connection=None, queue=None, business_key=None, labels=None, memo=None, search_attributes=None, duplicate_start_policy=None)

Command requesting a durable Nexus service operation from workflow code.

The Python worker executes the operation through the service-catalog API, records the response or typed failure as a side-effect marker, and then resumes replay from that recorded marker.

RecordVersionMarker `dataclass` ¶

RecordVersionMarker(change_id, version, min_supported, max_supported, result_kind='version')

Command recording a workflow code-version marker.

UpsertSearchAttributes `dataclass` ¶

UpsertSearchAttributes(attributes)

Command updating workflow search attributes.

UpsertMemo `dataclass` ¶

UpsertMemo(entries)

Merge non-indexed workflow memo metadata through durable history.

None deletes a key. The SDK encodes the complete patch in the public Avro payload envelope consumed by Server and Cloud runtimes.

WaitCondition `dataclass` ¶

WaitCondition(predicate, condition_key=None, condition_definition_fingerprint=None, timeout_seconds=None)

Command that yields execution until a workflow-defined predicate becomes true.

The replayer evaluates predicate locally against in-memory workflow state (typically mutated by signal/update handlers). The server records a ConditionWaitOpened history event and re-drives the workflow when any signal arrives or, if timeout_seconds is provided, when the timeout elapses (a TimerFired history event with timer_kind=condition_timeout).

SelectGroup `dataclass` ¶

SelectGroup(operations)

A durable first-completion group.

Yield this value to start every member and resume with a SelectionResult when Server commits the first eligible winner. Members that do not win remain durable and addressable through the returned handles.

DurableOperationHandle `dataclass` ¶

DurableOperationHandle(key, index, kind, identity, base_sequence, size, selection_group_id, operation)

Stable reference to one member of a durable selection group.

await_result ¶

await_result()

Return the yieldable handle used to await this member later.

cancel ¶

cancel()

Return the void/unit cancellation request yieldable.

Only SelectionOperationCancelled history proves cancellation won; awaiting a member that completed first still returns that completion.

SelectionResult `dataclass` ¶

SelectionResult(key, index, kind, identity, value, failure, winner, handles)

The one winner committed for a durable selection group.

CancelDurableOperation `dataclass` ¶

CancelDurableOperation(handle)

Explicitly cancel one still-running durable selection member.

MessageStreamMessage `dataclass` ¶

MessageStreamMessage(stream_name, message_id, position, arguments)

One server-ordered message consumed from a durable named stream.

MessageStream ¶

MessageStream(context, name)

Replay-safe receiver for repeated, ordered workflow input.

receive ¶

receive(max_items=1)

Wait for one message, then return a bounded currently-available batch.

receive_one ¶

receive_one()

Wait for and return the next message.

Saga ¶

Saga(context)

Deterministic reverse-order activity compensation helper.

Register a compensation only after its corresponding forward step has completed. run compensates on ordinary workflow failure or cooperative cancellation, using regular activity commands so replay uses the existing command/history protocol.

add_compensation ¶

add_compensation(activity_type, arguments=None, *, queue=None, retry_policy=None, start_to_close_timeout=None, schedule_to_start_timeout=None, schedule_to_close_timeout=None, heartbeat_timeout=None)

Register one compensation at the current deterministic position.

compensate ¶

compensate(initiating_failure)

Yield registered compensations in reverse order.

Compensation stops at the first compensation failure and raises SagaCompensationFailed, preserving both typed causes.

run ¶

run(forward)

Run forward and compensate if it fails or is cancelled.

WorkflowContext ¶

WorkflowContext(*, workflow_id='', run_id='', current_time=None, external_storage=None, external_storage_cache=None, workflow_command_id=None, cancel_requested=False)

Replay-safe helper surface passed to workflow run methods.

is_cancellation_requested `property` ¶

is_cancellation_requested

Whether this workflow task requests cooperative cancellation.

message_stream ¶

message_stream(name)

Open an instance-scoped durable input stream by its portable name.

throw_if_cancellation_requested ¶

throw_if_cancellation_requested()

Raise WorkflowCancelled at an explicit safe point.

saga ¶

saga()

Create a deterministic reverse-order compensation scope.

start_timer ¶

start_timer(seconds)

Yield a durable timer that resolves after seconds seconds.

sleep ¶

sleep(seconds)

Sleep for seconds seconds of durable wall time.

Sugar over start_timer that accepts a float and rounds up to the next whole second (the server stores timer deadlines as integer seconds). The call is still a single yield of a durable command use yield ctx.sleep(60) or bare yield ctx.sleep(60) from the workflow run method.

wait_condition ¶

wait_condition(predicate, *, key=None, timeout=None)

Yield execution until predicate() returns truthy.

The predicate is evaluated against the workflow's in-memory state on every replay tick typically mutated by @signal / @update handlers as external events arrive. If timeout is provided and elapses before the predicate becomes true, the yield resolves to False (otherwise True). The fractional timeout is rounded up to the next whole second to match the server's integer-second timer resolution.

select ¶

select(operations)

Start independent durable operations and wait for one winner.

A mapping preserves application-defined member keys. A sequence uses stable integer indexes. Supported members are activities, child workflows, timers, conditions, and nested ordinary parallel lists.

append_workflow_stream ¶

append_workflow_stream(stream_name, items, *, max_pending_items=None)

Append output items at a replay-safe workflow command boundary.

Each typed item accepts a payload value or explicit payload_reference plus optional item_type and content_type fields. The durable workflow command identity is used to derive stable per-item idempotency keys.

close_workflow_stream ¶

close_workflow_stream(stream_name, *, retention_seconds=None)

Close a run-scoped output stream at a replay-safe boundary.

error_workflow_stream ¶

error_workflow_stream(stream_name, error_reason, *, retention_seconds=None)

Mark a run-scoped output stream errored at a replay-safe boundary.

call_nexus_service ¶

call_nexus_service(endpoint_name, service_name, operation_name, arguments=None, *, idempotency_key=None, payload_codec=None, mode='sync', wait_for='completed', wait_timeout_seconds=None, caller_namespace=None, service_sdk_language=None, artifact_tuple=None, published_artifact_worker_execution=None, target_workflow_instance_id=None, target_workflow_run_id=None, connection=None, queue=None, business_key=None, labels=None, memo=None, search_attributes=None, duplicate_start_policy=None)

Yield a durable Nexus service operation and resume with its result.

If the service reports a typed failure, replay raises NexusOperationFailed at the yield point so workflow code can compensate or let the workflow fail.

start_nexus_operation ¶

start_nexus_operation(endpoint_name, service_name, operation_name, arguments=None, **kwargs)

Yield an async Nexus service operation and resume when it is accepted.

patched ¶

patched(change_id)

Record or read a patch marker and resolve to True for patched runs.

New runs record version 1 for change_id and replay as True. Older runs that reached this code without a marker resolve the legacy default version -1 and replay as False.

deprecate_patch ¶

deprecate_patch(change_id)

Keep a patch marker alive after the old branch has been removed.

upsert_memo ¶

upsert_memo(entries)

Return a replayable memo merge command with structural validation.

Replayer ¶

Replayer(*, workflows)

Replay captured workflow history without a live server.

Register one or more workflow classes, then call replay with a server-exported history list or a dictionary containing an events key. If the history includes a WorkflowStarted event, the replayer can infer the workflow type and start input from that event.

defn ¶

defn(*, name)

Register a class as a workflow type under a language-neutral name.

Scans the class for @signal, @query, and @update decorated methods and builds registries at decoration time so worker-side dispatch can use stable receiver names without re-inspecting the class on every history event or control-plane request.

signal ¶

signal(name)

Mark a workflow method as the handler for an external signal.

Example::

@workflow.defn(name="approval")
class ApprovalWorkflow:
    def __init__(self) -> None:
        self.approved: bool = False

    @workflow.signal("approve")
    def on_approve(self, by: str) -> None:
        self.approved = True

The decorated method is called by the replayer when a matching SignalReceived history event is observed, with the signal's decoded arguments unpacked into positional parameters. Handler return values are ignored; to expose state back to the workflow's main run loop, mutate self.* attributes (as on_approve does above) and yield the usual commands from run().

query ¶

query(name)

Mark a workflow method as a read-only query handler.

Query methods are invoked against replayed workflow state. They must not mutate self or perform I/O. The server-routed worker query path uses this decorator's receiver metadata and query_state to select and invoke the handler after replaying durable history.

Workers advertise query_tasks only after Server discovery reports worker_protocol.server_capabilities.query_tasks: true. Clients check the same capability before dispatch and raise a typed capability or discovery error when the query path is unavailable.

update ¶

update(name)

Mark a workflow method as an update handler.

The returned function also exposes .validator for the common pattern::

@workflow.update("approve")
def approve(self, approved: bool) -> dict: ...

@approve.validator
def validate_approve(self, approved: bool) -> None: ...

The worker advertises validator metadata during registration and refuses to register validator-bearing workflows against a server that cannot enforce synchronous validation before acceptance.

update_validator ¶

update_validator(name)

Mark a workflow method as the validator for an update name.

registry ¶

registry()

Return a copy of workflow types registered in this process.

commands_to_server_commands ¶

commands_to_server_commands(commands, task_queue, *, payload_codec=serializer.AVRO_CODEC, size_warning=serializer.DEFAULT_PAYLOAD_SIZE_WARNING, warning_context=None, external_storage=None, external_storage_threshold_bytes=None)

Convert workflow commands to the server wire shape with batched payload encoding.

query_state ¶

query_state(workflow_cls, history_events, start_input, query_name, args=None, *, workflow_id=None, run_id='', payload_codec=None, external_storage=None, external_storage_cache=None)

Replay a workflow to current state and invoke a registered query.

This is the replay and invocation core used by the current server-routed worker query path after it fetches durable history. That path is negotiated through worker_protocol.server_capabilities.query_tasks: workers advertise query_tasks only after discovery reports it as available, and clients fail with typed capability or discovery errors when it is not. Unknown query names and handler exceptions are normalized to QueryFailed.

apply_update ¶

apply_update(workflow_cls, history_events, start_input, update_id, *, workflow_id=None, run_id='', payload_codec=None, external_storage=None, external_storage_cache=None)

Replay current workflow state and run one accepted update handler.

The server remains the durable authority: it accepts the update, sends a workflow task carrying workflow_update_id, and records UpdateApplied / UpdateCompleted when this helper's worker command is submitted. Python only reconstructs in-memory state and runs the registered receiver method for the accepted update.

validate_update ¶

validate_update(workflow_cls, history_events, start_input, update_name, args, *, workflow_id=None, run_id='', payload_codec=None, external_storage=None, external_storage_cache=None)

Replay state and invoke only the declared pre-accept update validator.

The replayed instance is discarded after validation. This helper does not run the update handler and does not emit workflow commands, so the server remains the sole authority that can cross the accepted-state boundary.