dex

BOOL = _PrimitiveCodec(type_name='bool', wire_kind=<WireKind.BOOL: 'bool'>, expected_type=<class 'bool'>, validator=None)
BYTES = _PrimitiveCodec(type_name='bytes', wire_kind=<WireKind.BYTES: 'bytes'>, expected_type=<class 'bytes'>, validator=None)
DOUBLE = _PrimitiveCodec(type_name='float', wire_kind=<WireKind.DOUBLE: 'double'>, expected_type=<class 'float'>, validator=<function _validate_double>)
INT64 = _PrimitiveCodec(type_name='int', wire_kind=<WireKind.INT64: 'int64'>, expected_type=<class 'int'>, validator=<function _validate_int64>)
STRING = _PrimitiveCodec(type_name='str', wire_kind=<WireKind.STRING: 'string'>, expected_type=<class 'str'>, validator=None)
class ActiveStepSearchMode(enum.Enum):

Control which active Steps are included in Flow search indexing.

Attributes:
  • DEFAULT: Use the Dex server's current default policy.
  • ALL: Index every active Step.
  • WITH_WAIT_FOR: Index only active Steps that define wait_for.
  • DISABLED: Do not index active Steps.
DEFAULT = <ActiveStepSearchMode.DEFAULT: 'default'>
ALL = <ActiveStepSearchMode.ALL: 'all'>
WITH_WAIT_FOR = <ActiveStepSearchMode.WITH_WAIT_FOR: 'with_wait_for'>
DISABLED = <ActiveStepSearchMode.DISABLED: 'disabled'>
@dataclass(frozen=True)
class Attribute(typing.Generic[~ValueT]):

Define a typed, durable singleton value owned by a Flow.

Declare Attributes once in a Flow's PersistenceSchema and reuse the same definition from Steps, RPCs, and Client calls. Values are read and written through a handler Context; Client methods provide external access.

Set sync_to_attribute_store to project every write asynchronously through the Flow's configured Attribute Store. Projection is latest-state only, deletion projects SQL NULL, and failures do not roll back the Flow Attribute.

Set hard_delete_on_null=True for short-lived secrets, such as API keys, that must be removable. Dex stores every value of a hard-delete Attribute only in the Dex blob store and never in Flow history. Deleting the Attribute, or setting it to None, permanently deletes the stored value. A deleted value reads as absent, so get returns None.

A hard-delete Attribute accepts only string and object values, including bytes and dataclasses. int, bool, and float value types are rejected when the Attribute is defined. A custom codec that produces a numeric or Boolean value is rejected when its Flow is registered, and any numeric or Boolean value is rejected when it is written. The option cannot be combined with index or sync_to_attribute_store. Client.wait_for_attribute_match and AsyncClient.wait_for_attribute_match reject a hard-delete Attribute before contacting Dex. start_flow and SubFlow starts reject an initial value whose definition sets hard_delete_on_null differently from the target Flow's registered definition of that name, so a secret is never sent inline by mistake.

Hard-delete writes require a Dex Server with an enabled blob store. This SDK requires Server protocol 4. Workers negotiate before startup and Clients negotiate lazily before their first service request. Incompatible Servers fail with RuntimeError. A Server without an enabled blob store rejects the write. Step and RPC handlers receive the current value inline. The SDK never stores the value in the local blob cache.

Hard deletion has these limitations:

  • Each Flow keeps one stored copy per Attribute key. Every write overwrites that copy in place.
  • A Step or RPC attempt writes or deletes the stored value before its result is committed. A retried attempt, other Steps, and RPC handlers can therefore observe a write or deletion from an attempt that later failed.
  • When concurrent Steps or RPCs write the same key, the final stored value may not be the last committed write.
  • A failed start_flow or SubFlow start for a Flow ID that is already running still overwrites the running Flow's value. A None initial value deletes it.
  • Executions that reuse a Flow ID share the stored value.
  • Client.time_travel does not restore a deleted value. The value stays absent.
  • A Flow retry or a restarted SubFlow starts from the original initial value. If an earlier run deleted that value, the new run reads it as absent.
  • Removing hard_delete_on_null from a definition leaves the stored value in place until Dex removes the Flow. Delete the value with the hard-delete definition first.
  • Values remain until deleted, or until the Flow is removed by retention and blob cleanup.
  • Blob store backups or bucket versioning can retain copies outside Dex.
Attributes:
  • name: The non-empty logical Attribute name without /, unique within its Flow.
  • value_type: The Python type used to encode and decode values.
  • index: Optional search-index configuration; None disables indexing.
  • sync_to_attribute_store: Whether writes are projected to the selected Attribute Store. Defaults to False.
  • hard_delete_on_null: Whether values are stored only in the Dex blob store and permanently deleted when the Attribute is deleted or set to None. Defaults to False. Requires a string or object value_type and no index or sync_to_attribute_store.
Raises:
Examples:
>>> status = Attribute("status", str, sync_to_attribute_store=True)
>>> status.set(context, "paid")
>>> status.get(context)
'paid'

A Step uses a short-lived API key and then permanently deletes it:

>>> class CallProviderStep(Step[None]):
...     def __init__(self, api_key: Attribute[str]) -> None:
...         self.api_key = api_key
...
...     def execute(self, context: Context, input: None) -> StepDecision:
...         call_provider(self.api_key.get(context))
...         self.api_key.delete(context)
...         return graceful_complete("called")
>>> class ProviderFlow(Flow[None]):
...     def __init__(self) -> None:
...         self.api_key = Attribute("api-key", str, hard_delete_on_null=True)
...         self.call = CallProviderStep(self.api_key)
...
...     def get_steps(self) -> StepList[None]:
...         return StepList.start_step(self.call)
...
...     def get_persistence_schema(self) -> PersistenceSchema:
...         return PersistenceSchema.of(self.api_key)
>>> flow = ProviderFlow()
>>> with Client(Registry([flow]), cache) as client:
...     client.start_flow(
...         flow,
...         "provider-42",
...         None,
...         StartFlowOptions().with_attribute(flow.api_key, "sk-live-123"),
...     )
Attribute( name: str, value_type: type[~ValueT], index: AttributeIndex | None = None, sync_to_attribute_store: bool = False, hard_delete_on_null: bool = False)
name: str
value_type: type[~ValueT]
index: AttributeIndex | None = None
sync_to_attribute_store: bool = False
hard_delete_on_null: bool = False
def get(self, context: Context) -> ~ValueT:

Return the current value from a Step or RPC Context.

Arguments:
  • context: The current handler Context.
Returns:

The decoded value.

Raises:
  • KeyError: If the Attribute has no value.
  • TypeError: If the stored value cannot be decoded as value_type.
def set(self, context: Context, value: ~ValueT) -> None:

Stage a value to persist with the current handler decision.

Arguments:
  • context: The current handler Context.
  • value: A value compatible with value_type.
Raises:
  • TypeError: If value cannot be encoded as value_type.
def delete(self, context: Context) -> None:

Stage deletion of this Attribute in the current handler decision.

Arguments:
  • context: The current handler Context.
def lock(self) -> AttributeLock:

Return a lock request for this Attribute.

Use the result in Step or RPC lock options to serialize handlers that access the same value.

Returns:

A lock for this Attribute.

@dataclass(frozen=True)
class AttributeIndex:

Configure the search index for an Attribute or AttributeMap.

Attributes:
  • type: The value representation accepted by the search index.
  • index_key: The physical search key. An empty value uses the Attribute name for a singleton Attribute; AttributeMap definitions require an explicit key so all instances share one index.
AttributeIndex(type: IndexType, index_key: str = '')
type: IndexType
index_key: str = ''
@dataclass(frozen=True)
class AttributeLock:

Describe an Attribute lock acquired for a Step or RPC handler.

Attributes:
  • attribute: The singleton Attribute or AttributeMap definition to lock.
  • instance: The AttributeMap instance, or None for a singleton Attribute.
AttributeLock( attribute: Union[Attribute[Any], AttributeMap[Any]], instance: str | None = None)
attribute: Union[Attribute[Any], AttributeMap[Any]]
instance: str | None = None
@dataclass(frozen=True)
class AttributeMap(typing.Generic[~ValueT]):

Define a typed family of durable values keyed by map instance.

AttributeMap instances share one schema definition while keeping independent values and locks. Declare the map in PersistenceSchema before using it. Instance keys must be non-empty and must not contain /. Synced instances use their physical Attribute names as target columns. Projection is asynchronous and latest-state only, deletion projects SQL NULL, and failures do not roll back Flow Attributes.

Set hard_delete_on_null=True for short-lived secrets keyed by instance, such as one API key per provider. Every instance follows the hard-delete rules documented on Attribute. Values are stored only in the Dex blob store and never in Flow history. Deleting an instance, or setting it to None, permanently deletes its stored value. Only string and object value types, including bytes, are supported. The option cannot be combined with index or sync_to_attribute_store, and wait_for_attribute_match rejects it. Writes require a Server with an enabled blob store and Server protocol version 3 or later. Handlers receive values inline, and the SDK never stores them in the local blob cache. Each Flow keeps one stored copy per instance key. The retry, concurrency, failed-start, Flow ID reuse, time travel, retention, and backup limitations on Attribute apply to each instance.

Attributes:
  • name: The non-empty logical Attribute name without /, unique within its Flow.
  • value_type: The Python type used for every map instance.
  • index: Optional shared search-index configuration.
  • sync_to_attribute_store: Whether every instance is projected to the selected Attribute Store. Defaults to False.
  • hard_delete_on_null: Whether every instance value is stored only in the Dex blob store and permanently deleted when the instance is deleted or set to None. Defaults to False. Requires a string or object value_type and no index or sync_to_attribute_store.
Raises:
Examples:
>>> balances = AttributeMap("balance", int, sync_to_attribute_store=True)
>>> balances.set(context, "merchant-7", 1200)
>>> balances.get(context, "merchant-7")
1200

RPCs rotate and revoke one provider's API key:

>>> class WaitForRevocationStep(Step[None]):
...     def __init__(self, revoked: Channel[None]) -> None:
...         self.revoked = revoked
...
...     def wait_for(self, context: Context, input: None) -> Wait:
...         return Wait.until(self.revoked.for_one())
...
...     def execute(self, context: Context, input: None) -> StepDecision:
...         return graceful_complete("revoked")
>>> class CredentialFlow(Flow[None]):
...     def __init__(self) -> None:
...         self.api_keys = AttributeMap(
...             "provider-api-key", str, hard_delete_on_null=True
...         )
...         self.revoked = Channel("revoked", type(None))
...         self.wait = WaitForRevocationStep(self.revoked)
...
...     def get_steps(self) -> StepList[None]:
...         return StepList.start_step(self.wait)
...
...     def get_persistence_schema(self) -> PersistenceSchema:
...         return PersistenceSchema.of(self.api_keys, self.revoked)
...
...     @rpc
...     def rotate_api_key(self, context: Context, new_key: str) -> None:
...         self.api_keys.set(context, "payments", new_key)
...
...     @rpc
...     def revoke_api_key(self, context: Context) -> None:
...         self.api_keys.delete(context, "payments")
...         self.revoked.publish(context, None)
>>> flow = CredentialFlow()
>>> with Client(Registry([flow]), cache) as client:
...     client.start_flow(flow, "credentials-7", None)
...     client.invoke_rpc(flow.rotate_api_key, "credentials-7", "sk-live-456")
...     client.invoke_rpc(flow.revoke_api_key, "credentials-7")
AttributeMap( name: str, value_type: type[~ValueT], index: AttributeIndex | None = None, sync_to_attribute_store: bool = False, hard_delete_on_null: bool = False)
name: str
value_type: type[~ValueT]
index: AttributeIndex | None = None
sync_to_attribute_store: bool = False
hard_delete_on_null: bool = False
def load(self, instance: str) -> AttributeMapLoad:

Select one instance for an RPC snapshot.

Pass the result through @rpc(load_attribute_map_instances=(...)).

Arguments:
  • instance: The non-empty, slash-free logical map key to load.
Returns:

A load for this one instance.

Raises:
  • ValueError: If instance is empty or contains /.
def instance(self, instance: str) -> AttributeMapInstance[~ValueT]:

Select one typed instance for a Client Attribute read.

Pass the result to Client.get_attributes or AsyncClient.get_attributes, and use the same selector, or an equal one, to look up the value in the returned AttributeValues. The selector keeps the AttributeMap value type, so type checkers infer the value type of the lookup.

This differs from load(). load selects an instance that an RPC handler receives in its snapshot. instance selects an instance that a Client reads from outside the Flow.

Example::

notes = AttributeMap("notes", Note)
values = client.get_attributes("order-42", notes.instance("n1"))
note = values.get(notes.instance("n1"))
Arguments:
  • instance: The non-empty, slash-free logical map key to read.
Returns:

A typed selector for this one instance.

Raises:
  • TypeError: If instance is not a string.
  • ValueError: If instance is empty or contains /.
def get(self, context: Context, instance: str) -> ~ValueT:

Return one map instance from a Step or RPC Context.

Arguments:
  • context: The current handler Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
Returns:

The decoded instance value.

Raises:
  • KeyError: If the instance has no value.
  • ValueError: If instance is empty or contains /.
def set( self, context: Context, instance: str, value: ~ValueT) -> None:

Stage one map-instance value for the current handler decision.

Arguments:
  • context: The current handler Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • value: A value compatible with value_type.
def delete(self, context: Context, instance: str) -> None:

Stage deletion of one map instance.

Arguments:
  • context: The current handler Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
def get_map_size(self, context: Context) -> int:

Return the number of existing instances, including buffered writes.

Arguments:
  • context: The current Step or RPC Context.
Returns:

The number of keys visible after decision-local writes and deletions.

def get_all_instance_keys(self, context: Context) -> tuple[str, ...]:

Return decoded existing instance keys in ascending order.

Arguments:
  • context: The current Step or RPC Context.
Returns:

An immutable tuple reflecting decision-local writes and deletions.

def lock(self, instance: str) -> AttributeLock:

Return a lock request for one map instance.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
Returns:

A lock for instance.

Raises:
  • ValueError: If instance is empty or contains /.
@dataclass(frozen=True)
class AttributeMapInstance(typing.Generic[~ValueT]):

Select one typed AttributeMap instance for a Client Attribute read.

Create a selector with AttributeMap.instance(). Pass it to Client.get_attributes or AsyncClient.get_attributes together with singleton Attributes. Then use it as the key of the returned AttributeValues. Selectors are immutable and hashable. Two selectors are equal when their AttributeMap definitions are equal and their instance keys are the same string.

Example::

notes = AttributeMap("notes", Note)
first_note = notes.instance("n1")
values = client.get_attributes("order-42", first_note)
if first_note in values:
    print(values[first_note].text)
Attributes:
  • attribute_map: The AttributeMap definition that owns the instance.
  • instance: The non-empty logical map key. It must not contain /.
Raises:
AttributeMapInstance(attribute_map: AttributeMap[~ValueT], instance: str)
attribute_map: AttributeMap[~ValueT]
instance: str
@dataclass(frozen=True)
class AttributeMapLoad:

Load one AttributeMap instance for an RPC snapshot.

Create a load with AttributeMap.load(). Loading provides a point-in-time value snapshot; it does not lock the map against concurrent writers.

Attributes:
  • attribute_map: The exact AttributeMap definition registered with the Flow.
  • instance: One logical instance key.
AttributeMapLoad(attribute_map: AttributeMap[typing.Any], instance: str)
attribute_map: AttributeMap[typing.Any]
instance: str
physical_name: str

Return the physical instance name for this typed load.

Returns:

The encoded physical instance name.

class AttributeMapNotLoadedError(builtins.RuntimeError):

Indicate that an RPC did not load the requested AttributeMap entries.

@dataclass(frozen=True)
class AttributeMatch(typing.Generic[~ValueT]):

Describe a scalar Attribute predicate.

Create matches with the class methods and pass them to Client.wait_for_attribute_match() or AsyncClient.wait_for_attribute_match(). String and bool Attributes support equality operators. Int and float Attributes support every operator. A missing Attribute never matches.

Examples:
>>> match = AttributeMatch.greater_than(3)
AttributeMatch( _operator: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object>, _operand: ~ValueT)
@classmethod
def equal_to(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create an equality match.

Arguments:
  • operand: Value the Attribute must equal.
Returns:

A match requiring equality with operand.

@classmethod
def not_equal_to(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create an inequality match.

Arguments:
  • operand: Value the Attribute must differ from.
Returns:

A match requiring an existing value different from operand.

@classmethod
def greater_than(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create a numeric greater-than match.

Arguments:
  • operand: Numeric lower bound excluded from the match.
Returns:

A match requiring a value greater than operand.

@classmethod
def greater_than_or_equal(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create a numeric greater-than-or-equal match.

Arguments:
  • operand: Numeric lower bound included in the match.
Returns:

A match requiring a value greater than or equal to operand.

@classmethod
def less_than(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create a numeric less-than match.

Arguments:
  • operand: Numeric upper bound excluded from the match.
Returns:

A match requiring a value less than operand.

@classmethod
def less_than_or_equal(cls, operand: ~ValueT) -> AttributeMatch[~ValueT]:

Create a numeric less-than-or-equal match.

Arguments:
  • operand: Numeric upper bound included in the match.
Returns:

A match requiring a value less than or equal to operand.

class AttributeValues:

Hold the typed values returned by one get_attributes call.

Client.get_attributes and AsyncClient.get_attributes return an AttributeValues. Look up each value with the Attribute or AttributeMapInstance that was passed to the call. An equal definition or selector finds the same value. Type checkers infer the value type from the key, so values[ORDER_STATUS] has the Attribute's value type.

A read is set when the Flow run stores a value for it. A stored value that decodes to None is still set. A read is unset when the run stores no value for it. A key that was not part of the call behaves like an unset read:

  • values[key] returns the value of a set read and raises KeyError otherwise.
  • values.get(key) returns the value of a set read and None otherwise.
  • values.get(key, default) returns the value of a set read and default otherwise.
  • key in values is True only for a set read.

AttributeValues is read-only and not iterable. Keep the reads you passed to look up their values. Its representation and KeyError messages name each read by its physical key, such as notes/n1, but never show values, so they are safe to log for hard-delete Attributes. AttributeValues() creates an empty result in which every read is unset.

Example::

ORDER_STATUS = Attribute("order-status", str)
NOTES = AttributeMap("notes", Note)

values = client.get_attributes("order-42", ORDER_STATUS, NOTES.instance("n1"))
status = values.get(ORDER_STATUS, "new")
if NOTES.instance("n1") in values:
    print(values[NOTES.instance("n1")].text)
AttributeValues()

Create an empty result in which every read is unset.

def get( self, read: Union[Attribute[Any], AttributeMapInstance[Any]], default: object = None) -> Any:

Return the value of a read, or a default when it is unset.

Use get when an unset Attribute has an application default. Use key in values instead when a stored None must be told apart from an unset read.

Arguments:
  • read: The Attribute or AttributeMapInstance passed to get_attributes, or an equal one.
  • default: The value returned when the read is unset or was not part of the call. Defaults to None.
Returns:

The decoded value of a set read, or default.

class AsyncContext(dex.Context, typing.Protocol):

Expose Context operations available to asynchronous Step handlers.

AsyncWorker supplies this protocol to coroutine wait_for and execute methods. Stream writes enqueue synchronously; heartbeat waits only for the Worker output queue and never for Dex Stream Store persistence.

async def heartbeat(self, value: object = Ellipsis) -> None:

Emit one heartbeat from the current asynchronous Step attempt.

Calling this method without value clears prior persisted details. Passing any value, including Python None, persists a present Value. Local activity execution ignores heartbeat frames.

Arguments:
  • value: Optional codec-supported checkpoint value.
Raises:
  • asyncio.CancelledError: If the active Worker invocation is canceled.
  • ValueError: If called outside a Step handler.
  • ValueMappingError: If value cannot be encoded.
class AsyncClient:

Call Dex FlowService asynchronously through registered Flow definitions.

Methods perform non-blocking gRPC I/O and are safe to await from concurrent tasks. The Client owns its asynchronous channel but not its Registry or BlobCache; use async with or await close during application shutdown.

Construction performs no network I/O. The first service request negotiates Server protocol 4 with GetServerInfo. Service methods raise RuntimeError only when the Server reports a protocol range this SDK cannot use. The message names the Server version and says whether to upgrade the Server (server-zd newer than v0.1.16) or the SDK. That result is cached, and every later call raises it again. When GetServerInfo itself fails, for example because the Server is unreachable, the call raises the same DexServiceError as any other transport failure. Such failures are not cached, so the next call negotiates again. Negotiation counts against a call's request timeout.

Attributes:
  • registry: Flow definitions used to validate typed calls.
  • blob_cache: The open cache used to hydrate large values.
  • options: The effective ClientOptions.
Examples:
>>> async with AsyncClient(registry, cache) as client:
...     run_id = await client.start_flow(orders, "order-42", input)
...     result = (await client.wait_for_flow("order-42")).single_output(OrderResult)
AsyncClient( registry: Registry, blob_cache: BlobCache, options: ClientOptions | None = None)

Construct an AsyncClient and its lazy plaintext gRPC channel.

Arguments:
  • registry: The Flow Registry used for validation and codecs.
  • blob_cache: An open cache used for asynchronous hydration.
  • options: Service address and bearer credential; None uses defaults and sends no authorization metadata.
registry
blob_cache
options
async def start_flow( self, flow: Flow[~InputT], flow_id: str, input: ~InputT, options: StartFlowOptions = StartFlowOptions(timeout=None, timeout_policy=<FlowTimeoutPolicy.DEFAULT: 'default'>, timeout_handler_options=None, start_delay=None, id_reuse_policy=<IdReusePolicy.DEFAULT: 'default'>, retry_policy=None, _attribute_initializations=(), config_override=None, ignore_already_started=False, request_id=None)) -> str:

Start a Flow and return its server-assigned run ID.

The await completes after Dex accepts the Flow, not after completion. flow must be the exact registered instance and input must match its starting Step annotation.

The first request negotiates Server protocol 4 and caches compatibility. Construction performs no network I/O. An incompatible Server fails this and every later request; transient negotiation failures can be retried.

Arguments:
  • flow: The registered Flow instance to start.
  • flow_id: A non-empty application ID stable across runs.
  • input: The starting Step input, or None without a starting Step.
  • options: Timeout, reuse, retry, idempotency, and initial-state options.
Returns:

The server-assigned run ID.

Raises:
  • FlowAlreadyStartedError: If reuse policy rejects an existing Flow ID.
  • FlowDefinitionError: If flow is not registered.
  • ValueMappingError: If an input or initial Attribute cannot be encoded, including a numeric or Boolean value for a hard-delete Attribute.
  • ValueError: If an initial Attribute uses a different hard_delete_on_null setting than the Flow's registered definition of that name.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
  • DexServiceError: If FlowService rejects or cannot perform the request.
async def invoke_rpc( self, rpc_method: Callable[..., Any], flow_id: str, input: object = None, *, run_id: str = '', options: RPCInvokeOptions = RPCInvokeOptions(lock_attribute_map_instances=(), load_attribute_map_instances=(), load_channel_map_instances=())) -> Any:

Invoke a registered RPC and await its typed result.

Pass the bound method from the registered Flow instance. RPC timeout and Attribute locks come from @rpc configuration. A non-transactional RPC without Attribute locks starts from a backend query. If the handler returns no durable effects, a retained terminal execution can serve that query. Locks, transactional execution, returned effects, or server policy can require an active execution.

Arguments:
  • rpc_method: A bound method decorated with @rpc.
  • flow_id: The non-empty target Flow ID.
  • input: The annotated input, or None for an input-free RPC.
  • run_id: Optional exact run. With "", the server resolves the current execution.
  • options: Additive runtime AttributeMap locks and exact map-instance loads.
Returns:

The decoded RPCResult.output, or None for a no-output RPC.

Raises:
  • FlowDefinitionError: If the method is not registered.
  • RpcLockConflictError: If Attribute locks cannot be acquired.
  • WorkerInvocationError: If the application handler fails.
  • ValueMappingError: If input or output mapping fails.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or the selected path cannot use its closed execution.
  • DexServiceError: If the service call otherwise fails.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def write_stream( self, flow_id: str, stream: Stream[~ValueT], source: str, value: ~ValueT) -> None:

Await one typed best-effort Stream append with source metadata.

Arguments:
  • flow_id: Logical Flow instance ID; the Flow need not exist or be active.
  • stream: Exact Stream object registered in one Flow schema.
  • source: Non-empty informational source. Repeated values still append.
  • value: Typed message to append.
Raises:
  • ValueError: If the Flow ID or source is empty.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If the message cannot be encoded.
  • DexServiceError: If FlowService cannot append the message.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def read_stream( self, flow_id: str, stream: Stream[~ValueT], resume_token: str = '', timeout: datetime.timedelta | None = None) -> StreamMessage[~ValueT]:

Await the next retained Stream message after a resume token.

Arguments:
  • flow_id: Logical Flow instance ID used as the Stream instance key.
  • stream: Exact Stream object registered in one Flow schema.
  • resume_token: Previous message token, or empty for the retained head.
  • timeout: Optional non-negative server-side long-poll duration.
Returns:

The decoded message, next resume token, creation time, and source.

Raises:
  • LongPollTimeoutError: If no message arrives before the server wait expires.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If the retained message cannot be decoded.
  • DexServiceError: If FlowService cannot perform the read.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def list_stream_messages( self, flow_id: str, stream: Stream[~ValueT], page_size: int, before_page_token: str = '') -> StreamMessagesPage[~ValueT]:

Await one newest-first page of retained Stream messages without waiting for writes.

An empty token starts at the retained tail. Pass the returned page token unchanged to read older messages. Trimming may remove messages between pages.

Arguments:
  • flow_id: Logical Flow instance ID used as the Stream instance key.
  • stream: Exact Stream object registered in one Flow schema.
  • page_size: Positive maximum message count within the server-configured limit.
  • before_page_token: Opaque token from the preceding page, or "" first.
Returns:

Decoded messages in newest-first order and the next-page token.

Raises:
  • ValueError: If page_size is not positive.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If a retained message cannot be decoded.
  • DexServiceError: If FlowService cannot list the Stream.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def wait_for_flow( self, flow_id: str, timeout: datetime.timedelta | None = None) -> FlowResult:

Await Flow closure and return its terminal result.

timeout is a server-side long-poll duration, not a local task deadline. A LongPollTimeoutError is retryable by awaiting this method again. Parallel completion order is not deterministic; select results by Step identity.

Arguments:
  • flow_id: The non-empty target Flow ID.
  • timeout: Optional non-negative server-side wait duration.
Returns:

The terminal status, failure detail, and output-bearing completions.

Raises:
  • LongPollTimeoutError: If the Flow remains open when the wait expires.
  • FlowNotFoundError: If flow_id does not exist.
  • ValueMappingError: If Dex returns a malformed or incompatible hydrated value.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def stop_flow( self, flow_id: str, options: StopFlowOptions = StopFlowOptions(type=<StopType.CANCEL: 'cancel'>, reason=None)) -> None:

Request cancellation, termination, or failure of an active Flow.

The await completes after Dex accepts the request, not after terminal status.

Arguments:
  • flow_id: The non-empty target Flow ID.
  • options: Stop mode and optional recorded reason.
Raises:
  • FlowNotActiveOrNotFoundError: If no active run exists.
  • DexServiceError: If FlowService rejects or cannot perform the request.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def describe_flow(self, flow_id: str) -> FlowInfo:

Return summary metadata for the current or latest Flow run.

Arguments:
  • flow_id: The non-empty Flow ID to describe.
Returns:

Flow ID, run ID, Flow type, status, and UTC start time.

Raises:
  • FlowNotFoundError: If flow_id does not exist.
  • DexServiceError: If FlowService cannot return the summary.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def get_attributes( self, flow_id: str, *reads: Union[Attribute[Any], AttributeMapInstance[Any]]) -> AttributeValues:

Read committed Attribute values from the current or latest Flow run.

Use get_attributes to read Flow state from outside the Flow without defining an RPC. Pass one or more reads. A read is a singleton Attribute or one AttributeMap instance selected with AttributeMap.instance(key). To read one Attribute, pass a single read. The Client sends every key in one FlowService request, so all values come from the same committed state. No Worker runs.

The call targets the current run, or the latest run when no run is open. A closed run stays readable until Dex purges it. The call sees only committed state. Unlike reading inside an RPC handler, it never sees writes that a running Step or RPC has staged but not yet committed.

Each read must use a definition that some Flow in this Client's Registry declares in its PersistenceSchema. The definition matches when it is equal to the registered one: same class (Attribute or AttributeMap), name, value_type, index, sync_to_attribute_store, and hard_delete_on_null. The definition may come from any registered Flow. The server does not check that the target Flow's type declares it, so a same-named Attribute of another type in the target Flow fails to decode.

Attributes defined with hard_delete_on_null=True are readable; the server resolves their stored values. The Client resolves other large values through LoadBlobs and the Client's BlobCache, and then decodes every value. On any error the call returns nothing, so no partial result is visible. Every error message names the Flow ID. An error about one read or value also names its Attribute name or physical key.

Example::

ORDER_STATUS = Attribute("order-status", str)
NOTES = AttributeMap("notes", Note)

try:
    values = await client.get_attributes(
        "order-42", ORDER_STATUS, NOTES.instance("n1")
    )
except FlowNotFoundError:
    return None
status = values.get(ORDER_STATUS, "new")
if NOTES.instance("n1") in values:
    note = values[NOTES.instance("n1")]
Arguments:
  • flow_id: The non-empty Flow ID.
  • *reads: At least one Attribute or AttributeMapInstance. Two reads of the same Attribute or AttributeMap instance are rejected.
Returns:

The set values, keyed by the reads passed to this call. An Attribute or AttributeMap instance without a stored value is absent from the result.

Raises:
  • ValueError: If flow_id is empty, no read is given, two reads select the same Attribute or AttributeMap instance, or the response has an unexpected key, a duplicate key, or an entry without a value.
  • TypeError: If a read is not an Attribute or AttributeMapInstance, such as an AttributeMap without an instance.
  • FlowDefinitionError: If no Flow in the Registry declares a matching definition, including a same-named definition with another value type.
  • FlowNotFoundError: If the Flow does not exist or its latest run is being purged.
  • ValueMappingError: If a returned value cannot be resolved or decoded as its value type. This includes an empty or non-finite value, an unsupported object encoding, and a large value that LoadBlobs omits.
  • DexServiceError: If FlowService or LoadBlobs otherwise fails. A LoadBlobs failure for a missing Flow raises FlowNotFoundError.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def search_flows( self, query: str, page_size: int, next_page_token: str = '', *, include_continued_as_new: bool = False) -> SearchFlowsPage:

Return one page of Flow runs matching a visibility query.

Arguments:
  • query: A Dex visibility query; an empty string uses server defaults.
  • page_size: The non-negative maximum result count requested.
  • next_page_token: Opaque token from the preceding page, or "" first.
  • include_continued_as_new: Include earlier runs without overriding query filters.
Returns:

Server-ordered entries with hydrated indexed Attributes and a next token.

Raises:
  • ValueError: If page_size is negative.
  • ValueMappingError: If an indexed Attribute cannot be hydrated.
  • DexServiceError: If FlowService cannot execute the query.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def time_travel(self, flow_id: str, options: TimeTravelOptions) -> str:

Create a new run from a selected point in existing Flow history.

Arguments:
  • flow_id: The non-empty Flow ID whose history supplies the new run.
  • options: Time travel selector, reason, and replay controls.
Returns:

The new server-assigned run ID; the Flow ID is unchanged.

Raises:
  • FlowNotFoundError: If flow_id does not exist.
  • ValueError: If time travel selector fields are invalid.
  • DexServiceError: If FlowService cannot create the new run.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def skip_timer( self, flow_id: str, step_execution_id: StepExecutionId, timer_id: TimerId) -> None:

Make one waiting Timer condition immediately ready.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • step_execution_id: The Step type and positive execution number.
  • timer_id: Exactly one Timer condition ID or zero-based index.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • ValueError: If an identifier is invalid.
  • DexServiceError: If FlowService cannot skip the Timer.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def wait_for_step_completion( self, flow_id: str, step_execution_id: StepExecutionId, options: WaitForStepCompletionOptions) -> None:

Await one Step execution completion or its request timeout expiry.

The server derives a stable Request ID from the Step execution when none is supplied. Request timeout covers transparent transport reattachments. Internal handler timeout controls Update generation rollover without ending this request.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • step_execution_id: The Step type and positive execution number.
  • options: The optional Request ID, request timeout, and internal handler timeout controls.
Raises:
  • ValueError: If the Request ID or request timeout is invalid.
  • RequestTimeoutError: If a positive request timeout expires first.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closes first.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def wait_for_attribute_match( self, flow_id: str, attribute: Union[Attribute[Any], AttributeMap[Any]], *args: object, **kwargs: object) -> Any:

Await a singleton Attribute or AttributeMap instance satisfying a match.

Singleton form is wait_for_attribute_match(flow_id, attribute, match, options); map form adds instance before match.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • attribute: The registered Attribute or AttributeMap to observe.
  • *args: Positional match and options, optionally preceded by a map instance.
  • **kwargs: The same arguments supplied by name.
Returns:

The current value that satisfied the match.

Raises:
  • TypeError: If arguments do not match the Attribute definition.
  • ValueError: If an identifier, option, or expected value is invalid. Also raised before transport when the Attribute uses hard_delete_on_null.
  • RequestTimeoutError: If a positive request timeout expires first.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closes first.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def update_flow_config(self, flow_id: str, config: FlowConfig) -> None:

Update specified mutable configuration fields for an active Flow.

The update affects later decisions and does not recall work already dispatched. Unspecified fields and the existing Worker target remain unchanged. An empty FlowConfig() leaves all configuration unchanged.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • config: New optional fields applied to later decisions.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • ValueError: If a configuration value is invalid.
  • DexServiceError: If FlowService cannot apply the update.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def trigger_continue_as_new(self, flow_id: str) -> None:

Ask an active Flow to roll history into a new run.

The Flow ID remains stable and the run ID changes when rollover occurs.

Arguments:
  • flow_id: The non-empty active Flow ID.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • DexServiceError: If FlowService cannot accept the request.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def health_check(self) -> bool:

Await the FlowService health endpoint.

Returns:

True when the service returns successfully.

Raises:
  • DexServiceError: If the service is unhealthy or unreachable.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
async def close(self) -> None:

Close the owned asynchronous gRPC channel.

close is idempotent. Other methods must not be awaited afterward. The Registry and BlobCache remain owned by the caller.

class AsyncBufferedTextStream:

Batch text chunks for an asynchronous Step invocation.

Create this writer with Stream.buffered_text(). write is synchronous and returns after local buffering or Worker-output enqueueing. AsyncWorker automatically stops its timer and flushes remaining text before the invocation result or error.

AsyncBufferedTextStream( stream: Stream[str], context: Context, flush_interval: datetime.timedelta, max_buffered_bytes: int)

Initialize an invocation-managed writer from validated factory inputs.

Arguments:
  • stream: Registered string Stream receiving each batch.
  • context: Current asynchronous Step Context.
  • flush_interval: Positive one-shot timer interval.
  • max_buffered_bytes: Positive soft UTF-8 batch threshold.
def write(self, chunk: str) -> None:

Append one text chunk and flush after reaching the soft size threshold.

Empty chunks are ignored. The first non-empty chunk starts the configured one-shot timer. This method does not wait for the timer or Stream Store acknowledgement.

Arguments:
  • chunk: Text to append without modification.
Raises:
  • TypeError: If chunk is not a string.
  • BaseException: The previously latched timer or Worker-output failure.
  • ValueError: If the invocation already finished.
class AsyncWorker:

Host asynchronous Step and RPC handlers through outbound pull streams.

A Worker is one-shot. Await start in a background task because it waits for termination, and await stop or close during shutdown. Workers need only an outbound connection to the Dex Server. Application handlers may overlap on the event loop and must avoid blocking operations.

Attributes:
  • registry: The Flow Registry served by this Worker.
  • blob_cache: The shared cache used to hydrate large values.
  • options: The effective WorkerOptions.
Examples:
>>> async with AsyncWorker(registry, cache) as worker:
...     task = asyncio.create_task(worker.start())
...     await worker.stop()
...     await task
AsyncWorker( registry: Registry, blob_cache: BlobCache, options: WorkerOptions | None = None)

Construct an AsyncWorker without starting its outbound pull streams.

Arguments:
  • registry: A Registry created with allow_async_handlers=True.
  • blob_cache: An open cache shared for asynchronous value hydration.
  • options: Networking, startup, and bearer credential options; None uses defaults and sends no authorization metadata.
Raises:
  • ValueError: If the pull Worker ID is invalid.
registry
blob_cache
options
async def start(self) -> None:

Negotiate protocol, synchronize Attribute indexes, and serve.

start may be awaited exactly once. It contacts FlowService before opening outbound task streams and completes only after stop terminates those streams. Permanent pull stream rejections raise PullRegistrationError when stop_on_permanent_stream_rejection is enabled; otherwise streams retry.

Raises:
  • PullRegistrationError: A permanent stream rejection stops this Worker.
  • RuntimeError: If lifecycle state is invalid, Server protocol 4 is unsupported, concurrency exceeds matching.maxConcurrentTasksCap, or index synchronization fails.
async def wait_pull_registered(self, timeout: float | None = None) -> None:

Wait until the required execution slot kinds have registered.

Start the Worker separately before waiting, or start it concurrently. Readiness requires one STEP_AND_RPC stream and, when RPC reservation is configured, one RPC_ONLY stream. Successful registration latches after stop. Use pull_registration_state for subsequent readiness probes. A permanent rejection before readiness raises PullRegistrationError only when no other stream is registered for that slot kind. Startup failures wake waiters. When the first registration reports a Server per-Worker limit below max_concurrent_tasks, every waiter raises that limit error and the Worker never becomes ready. Multiple callers may wait independently, including before start.

Arguments:
  • timeout: Maximum seconds to wait; None waits indefinitely.
Raises:
  • PullRegistrationError: A slot kind is permanently rejected before readiness.
  • TimeoutError: The timeout expires.
  • asyncio.CancelledError: The calling task is canceled; the Worker keeps running.
  • RuntimeError: The Worker stops before readiness, or max_concurrent_tasks exceeds the Server's matching.maxConcurrentTasksCap.
  • BaseException: The original startup failure, if start fails.
Examples:
>>> start_task = asyncio.create_task(worker.start())

>>> await worker.wait_pull_registered(timeout=30)
>>> is_ready = worker.pull_registration_state().is_ready
pull_reserved_rpc_declines: int

Return the deprecated reserved RPC decline counter, always zero.

Reserved RPC capacity uses RPC_ONLY slots. Steps are never declined for capacity. This property remains available for source compatibility and is safe to read concurrently, including after stop.

Returns:

Zero.

Examples:
>>> declines = worker.pull_reserved_rpc_declines
def pull_registration_state(self) -> PullRegistrationState:

Return an immutable snapshot of execution slot kind registrations.

Transient failures, reconnects, and stop preserve initial readiness. A permanent rejection makes readiness false only while that slot kind has no registered stream. Re-registration clears its error. Calls are safe concurrently with start, stop, and other waiters. See wait_pull_registered for an example.

Returns:

Current state grouped by STEP_AND_RPC and, when configured, RPC_ONLY.

async def stop(self) -> None:

Drain handlers and release asynchronous channel resources.

Calls before start and repeated calls after shutdown are safe. Pull mode drains for PullWorkerOptions.stop_drain_timeout (30 seconds by default), then cancels active tasks and waits for handlers to return. Idle streams half-close immediately. Busy streams send final output before half-closing. Tasks received after stop begins are dropped without running handlers. A timed-out busy stream is reset before cancellation. Half-closed streams keep reading server frames until EOF or the remaining drain deadline before the channel closes. When the deadline expires, every stream that is still open is reset, even one blocked on a send or without Server EOF, so stop returns once handlers return. The Server may count a task it sent in that window as an attempt that may have run. One exception applies to a stream whose final output was queued or written but not yet confirmed by Server EOF or by the next assignment. Such a stream is not reset at the drain deadline, even if its handler finished or it already half-closed. It gets up to 5 more seconds to send the output, half-close, and receive Server EOF, and is reset after that. Stop can therefore return up to 5 seconds after the drain deadline.

async def close(self) -> None:

Stop the AsyncWorker and permanently mark it closed.

Uses the same shutdown contract as stop: idle streams half-close, busy streams send final output before half-closing, and later tasks are dropped. A handler exceeding the drain timeout has its stream reset. Pull streams finish their bounded EOF wait before the channel closes; a stream still open at the deadline is reset. Repeated calls are safe; a closed Worker cannot be started again.

class ApiKeyCredential:

Exchange an API key for short-lived bearer tokens before Dex calls.

Use this when an application holds a long-lived API key and a token endpoint issues access tokens for a Dex server ID. Use FileCredential when another process writes and rotates a bearer token file instead. The first token() call exchanges synchronously. Later calls start one background refresh after 80% of the token lifetime and keep serving the valid token. Concurrent callers share one exchange. Failed refreshes retry no more often than every five seconds while the old token remains valid. An expired token makes callers wait up to exchange_timeout for the in-flight exchange. Dex unary calls retry once with a fresh token after UNAUTHENTICATED. Pull Worker streams reconnect once immediately with a fresh token after UNAUTHENTICATED; a second rejection uses the normal backoff.

The token URL must use HTTPS. HTTP is allowed only on localhost, 127.0.0.0/8, or ::1. Loopback HTTP bypasses environment proxies; HTTPS honors them. Redirects are refused. The instance is thread-safe. close() is optional and can be called after all Clients and Workers sharing the credential are closed. It waits only through the current exchange deadline. Formatting this credential or an exchange error does not reveal secrets.

Example:
>>> credential = ApiKeyCredential(
...     "https://auth.example.com/token", "dex-server", "application-api-key"
... )
>>> client = Client(registry, cache, ClientOptions(credential=credential))
>>> try:
...     client.health_check()
... finally:
...     client.close()
...     credential.close()
Arguments:
  • token_url: Absolute token endpoint URL.
  • server_id: Server ID used to scope the issued access token.
  • api_key: API key exchanged for access tokens.
  • exchange_timeout: Maximum seconds for each exchange, including connect and response; defaults to 10 seconds.
Raises:
  • ValueError: An input is empty, the URL is invalid or insecure, or the timeout is not positive.
ApiKeyCredential( token_url: str, server_id: str, api_key: str, *, exchange_timeout: float = 10.0)

Validate exchange settings without contacting the token endpoint.

Arguments:
  • token_url: Absolute HTTPS URL, or loopback HTTP URL.
  • server_id: Server ID to send in the exchange request.
  • api_key: API key to exchange for an access token.
  • exchange_timeout: Positive maximum exchange time in seconds.
Raises:
  • ValueError: A required string is empty, the URL is invalid or insecure, or the timeout is not positive and finite.
def token(self) -> str:

Return an access token, exchanging or refreshing it when necessary.

A valid token returns immediately, including during background refresh. First and expired-token calls wait up to exchange_timeout. Concurrent calls share the same exchange. Exchange failures raise ApiKeyExchangeError. Calls after close() raise RuntimeError.

Returns:

Access token without the Bearer prefix.

Raises:
  • ApiKeyExchangeError: The endpoint, response, or transport failed.
  • RuntimeError: The credential is closed.
def close(self) -> None:

Stop using this credential and wait up to its current exchange deadline.

Closing is optional. Close a shared credential after all Clients and Workers using it have closed. Subsequent token() calls raise RuntimeError. A network operation may continue in a daemon thread after the deadline when the operating system cannot interrupt it.

class ApiKeyExchangeError(builtins.Exception):

Report a failed API key exchange without exposing credential secrets.

The message identifies the token URL and HTTP status when available. It includes the endpoint's error and error_description fields, each limited to 200 characters. It never includes the response body, API key, issued access tokens, or JWT-shaped strings. Client calls raise this error unchanged. Pull Workers log it and retry.

class BlobCache(typing.Protocol):

Provide local, bounded storage for content-addressed Dex blobs.

Implementations may be called concurrently by Clients and Workers. Blob IDs are opaque server identifiers; callers should not derive their own IDs.

BlobCache(*args, **kwargs)
config: BlobCacheConfig

Return the configuration used to open this cache.

Returns:

The effective cache configuration.

def get(self, blob_id: str) -> bytes | None:

Read a cached payload without contacting Dex.

Arguments:
  • blob_id: The non-empty, opaque blob identifier.
Returns:

The payload bytes, or None when the entry is absent.

def put(self, blob_id: str, payload: bytes) -> bool:

Offer a payload to the bounded cache.

Arguments:
  • blob_id: The non-empty, opaque blob identifier.
  • payload: The bytes to retain; the cache does not take mutable ownership.
Returns:

True when admitted, or False when the policy rejects the entry.

def delete(self, blob_id: str) -> None:

Delete one cached payload if present.

Arguments:
  • blob_id: The opaque identifier to remove.
def delete_all(self) -> None:

Delete every cached payload while keeping the cache open.

def close(self) -> None:

Flush metadata and release native cache resources.

Repeated calls are safe; other methods must not be used after close.

@dataclass(frozen=True)
class BlobCacheConfig:

Configure the local persistent cache for large Dex values.

Attributes:
  • directory: The writable directory that stores cache data and metadata.
  • max_bytes: The positive maximum on-disk payload size in bytes.
  • frequency_counters: Admission-policy counter count. Zero selects the default of 10,000; larger values improve frequency estimates at a memory cost.
BlobCacheConfig(directory: str, max_bytes: int, frequency_counters: int = 10000)
directory: str
max_bytes: int
frequency_counters: int = 10000
class BufferedTextStream:

Batch text chunks for a synchronous Step generator.

Create this writer with Stream.buffered_text(). Yield every output returned by write and flush. The interval is cooperative: elapsed time is checked when another chunk arrives.

BufferedTextStream( stream: Stream[str], context: Context, flush_interval: datetime.timedelta, max_buffered_bytes: int)

Initialize a cooperative writer from validated factory inputs.

Arguments:
  • stream: Registered string Stream receiving each batch.
  • context: Current synchronous Step Context.
  • flush_interval: Positive cooperative flush interval.
  • max_buffered_bytes: Positive soft UTF-8 batch threshold.
def write(self, chunk: str) -> tuple[StepOutput, ...]:

Append one chunk and return an output when a threshold is reached.

Arguments:
  • chunk: Text to append without modification.
Returns:

An empty tuple while buffering, or one StepOutput that the handler must yield.

Raises:
  • TypeError: If chunk is not a string.
  • ValueMappingError: If the combined text cannot be encoded.
def flush(self) -> tuple[StepOutput, ...]:

Return the current batch as one StepOutput for the handler to yield.

Returns:

An empty tuple for an empty buffer, or one Stream StepOutput.

Raises:
  • ValueMappingError: If the combined text cannot be encoded.
@dataclass(frozen=True)
class Channel(typing.Generic[~ValueT]):

Define a typed, durable singleton message stream owned by a Flow.

Add the Channel to the Flow's PersistenceSchema. Publishers append values; Steps use the condition factories to wait for a count and then inspect the selected values through results.

Attributes:
  • name: The non-empty Channel name without /, unique within its Flow.
  • value_type: The Python type of every published value.
Examples:
>>> approvals = Channel("approvals", str)
>>> wait = Wait.until(approvals.for_n(2, condition_id="two-approvals"))
>>> approvals.publish(context, "reviewer@example.com")
Channel(name: str, value_type: type[~ValueT])
name: str
value_type: type[~ValueT]
def publish(self, context: Context, value: ~ValueT) -> None:

Stage one value to append with the current handler decision.

Arguments:
  • context: The current Step or RPC Context.
  • value: A value compatible with value_type.
def delete(self, context: Context, message_id: str) -> None:

Stage deletion of one pending message from an RPC handler.

Use @rpc(is_transactional=True) when a missing message must fail the entire RPC without committing its other writes. Step Contexts reject this operation.

Arguments:
  • context: The current RPC Context.
  • message_id: Non-empty ID returned by a Client pending-message read.
def size(self, context: Context) -> int:

Return the current number of queued values.

Arguments:
  • context: The current Step or RPC Context.
Returns:

The non-negative number of queued Channel values.

def pending_messages( self, context: Context) -> Sequence[ChannelMessage[~ValueT]]:

Return the pending messages loaded for the current RPC snapshot.

Configure the RPC with load_channels=(this_channel,). The returned sequence preserves FIFO order and does not change after staged publications or deletions.

Arguments:
  • context: The current RPC Context.
Returns:

Pending message IDs and decoded values in FIFO order.

Raises:
  • ChannelMessagesNotLoadedError: If the RPC did not load this Channel's messages.
def find_pending_message( self, context: Context, message_id: str) -> Optional[ChannelMessage[~ValueT]]:

Find one loaded pending message by server-assigned ID.

Arguments:
  • context: The current RPC Context.
  • message_id: The message ID to find.
Returns:

The matching message, or None when absent from the snapshot.

def results(self, context: Context) -> Sequence[~ValueT]:

Return values selected by the Step's satisfied Channel condition.

Arguments:
  • context: The current Step Context.
Returns:

An ordered, read-only sequence for this Step execution; it is empty when the Step did not wait on this Channel.

def for_one(self, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a condition that consumes exactly one value.

Arguments:
  • condition_id: Optional stable identifier used by Timer and wait APIs.
Returns:

A Channel condition equivalent to for_range(at_least=1, at_most=1).

def for_n( self, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a condition that consumes exactly count values.

Arguments:
  • count: The required non-negative value count.
  • condition_id: Optional stable condition identifier.
Returns:

A Channel condition with equal lower and upper bounds.

def at_least( self, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a condition requiring at least count values.

Arguments:
  • count: The inclusive, non-negative lower bound.
  • condition_id: Optional stable condition identifier.
Returns:

A Channel condition with no upper bound.

def at_most( self, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a non-blocking condition consuming up to count queued values.

The condition does not wait on its own for values to accumulate. When its surrounding Wait completes, Dex consumes up to count values queued at that time. An empty queue produces no values.

Arguments:
  • count: The inclusive, non-negative upper bound.
  • condition_id: Optional stable condition identifier.
Returns:

A Channel condition with no positive lower bound.

def for_range( self, *, at_least: int | None = None, at_most: int | None = None, condition_id: str | None = None) -> dex.condition.Condition:

Create a bounded condition for queued Channel values.

At least one bound is required. When both are present, at_least cannot exceed at_most. Dex waits only for at_least. Once that bound is met, it consumes currently queued values up to at_most. Omitting at_least makes the condition complete immediately.

Arguments:
  • at_least: Optional inclusive lower bound.
  • at_most: Optional inclusive upper bound.
  • condition_id: Optional stable identifier unique within the Wait tree.
Returns:

A condition accepted by Wait factories.

@dataclass(frozen=True)
class ChannelMap(typing.Generic[~ValueT]):

Define keyed Channel instances that share one typed schema.

Each instance has an independent queue and may be waited on separately. Add the ChannelMap definition to the Flow's PersistenceSchema. Instance keys must be non-empty and must not contain /.

Attributes:
  • name: The non-empty Channel name without /, unique within its Flow.
  • value_type: The Python type of every published value.
Examples:
>>> events = ChannelMap("events", dict[str, str])
>>> events.publish(context, "customer-42", {"kind": "paid"})
>>> wait = Wait.until(events.for_one("customer-42"))
ChannelMap(name: str, value_type: type[~ValueT])
name: str
value_type: type[~ValueT]
def load_messages(self, instance: str) -> ChannelMapLoad:

Select one instance's pending messages for an RPC snapshot.

Arguments:
  • instance: The non-empty, slash-free logical ChannelMap instance key.
Returns:

A load for this one instance.

Raises:
  • ValueError: If instance is empty or contains /.
def publish( self, context: Context, instance: str, value: ~ValueT) -> None:

Stage a value to append to one ChannelMap instance.

Arguments:
  • context: The current Step or RPC Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • value: A value compatible with value_type.
def delete( self, context: Context, instance: str, message_id: str) -> None:

Stage deletion of one pending message from a ChannelMap instance.

Arguments:
  • context: The current RPC Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • message_id: Non-empty ID returned by a Client pending-message read.
def size(self, context: Context, instance: str) -> int:

Return the queued value count for one instance.

Arguments:
  • context: The current Step or RPC Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
Returns:

The non-negative number of queued values for instance.

def pending_messages( self, context: Context, instance: str) -> Sequence[ChannelMessage[~ValueT]]:

Return one instance's pending messages from the loaded RPC snapshot.

Configure the RPC with load_channel_map_instances=(this_channel_map.load_messages(instance),) or load the whole map with load_channel_maps=(this_channel_map,).

Arguments:
  • context: The current RPC Context.
  • instance: The non-empty logical map key.
Returns:

Pending message IDs and decoded values in FIFO order.

Raises:
  • ChannelMessagesNotLoadedError: If the RPC did not load this ChannelMap's messages.
def find_pending_message( self, context: Context, instance: str, message_id: str) -> Optional[ChannelMessage[~ValueT]]:

Find one loaded ChannelMap message by server-assigned ID.

Arguments:
  • context: The current RPC Context.
  • instance: The non-empty logical map key.
  • message_id: The message ID to find.
Returns:

The matching message, or None when absent from the snapshot.

def get_map_size(self, context: Context) -> int:

Return the number of non-empty instances visible to the current RPC.

Arguments:
  • context: The current RPC Context.
Returns:

The number of keys after including publications buffered by this RPC.

def get_all_instance_keys(self, context: Context) -> tuple[str, ...]:

Return decoded non-empty instance keys in ascending order.

Arguments:
  • context: The current RPC Context.
Returns:

An immutable tuple including publications buffered by this RPC.

def results(self, context: Context, instance: str) -> Sequence[~ValueT]:

Return values selected for one instance by the satisfied condition.

Arguments:
  • context: The current Step Context.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
Returns:

An ordered, read-only sequence for this Step execution.

def for_one( self, instance: str, *, condition_id: str | None = None) -> dex.condition.Condition:

Create an instance condition that consumes exactly one value.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • condition_id: Optional stable condition identifier.
Returns:

A ChannelMap condition for one value.

def for_n( self, instance: str, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create an instance condition that consumes exactly count values.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • count: The required non-negative value count.
  • condition_id: Optional stable condition identifier.
Returns:

A condition with equal lower and upper bounds.

def at_least( self, instance: str, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create an instance condition requiring at least count values.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • count: The inclusive, non-negative lower bound.
  • condition_id: Optional stable condition identifier.
Returns:

A condition with no upper bound.

def at_most( self, instance: str, count: int, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a non-blocking instance condition consuming up to count values.

The condition does not wait on its own for values to accumulate. When its surrounding Wait completes, Dex consumes up to count values queued for the instance at that time. An empty queue produces no values.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • count: The inclusive, non-negative upper bound.
  • condition_id: Optional stable condition identifier.
Returns:

A condition with no positive lower bound.

def for_range( self, instance: str, *, at_least: int | None = None, at_most: int | None = None, condition_id: str | None = None) -> dex.condition.Condition:

Create a bounded condition for one ChannelMap instance.

Dex waits only for at_least. Once that bound is met, it consumes currently queued values up to at_most. Omitting at_least makes the condition complete immediately.

Arguments:
  • instance: The map instance. Slash is prohibited because it is a reserved character.
  • at_least: Optional inclusive lower bound.
  • at_most: Optional inclusive upper bound.
  • condition_id: Optional stable identifier unique within the Wait tree.
Returns:

A condition accepted by Wait factories.

@dataclass(frozen=True)
class ChannelMapLoad:

Load one ChannelMap instance's pending messages for an RPC snapshot.

Attributes:
  • channel_map: The exact ChannelMap definition registered with the Flow.
  • instance: One logical instance key.
ChannelMapLoad(channel_map: ChannelMap[typing.Any], instance: str)
channel_map: ChannelMap[typing.Any]
instance: str
physical_name: str

Return the physical instance name for this typed load.

Returns:

The encoded physical instance name.

@dataclass(frozen=True)
class ChannelMessage(typing.Generic[~ValueT]):

Represent one pending Channel message and its server-assigned identity.

Attributes:
  • message_id: UUIDv7 assigned by Dex when the message was published.
  • value: The decoded Channel value.
ChannelMessage(message_id: str, value: ~ValueT)
message_id: str
value: ~ValueT
class ChannelMessageNotFoundError(dex.DexServiceError):

Indicate that a pending Channel message ID no longer exists.

class ChannelMessagesNotLoadedError(builtins.RuntimeError):

Indicate that an RPC did not load the requested pending Channel messages.

class Client:

Call Dex FlowService through registered, typed Flow definitions.

Client methods are synchronous and may block on network I/O. Create one Client per Registry, share it across calling threads, and close it at application shutdown. The Client owns its gRPC channel but not the Registry or BlobCache.

Construction performs no network I/O. The first service request negotiates Server protocol 4 with GetServerInfo. Service methods raise RuntimeError only when the Server reports a protocol range this SDK cannot use. The message names the Server version and says whether to upgrade the Server (server-zd newer than v0.1.16) or the SDK. That result is cached, and every later call raises it again. When GetServerInfo itself fails, for example because the Server is unreachable, the call raises the same DexServiceError as any other transport failure. Such failures are not cached, so the next call negotiates again. Negotiation counts against a call's request timeout.

Attributes:
  • registry: Flow definitions used to validate typed calls.
  • blob_cache: The open cache used to hydrate large values.
  • options: The effective ClientOptions.
Examples:
>>> with Client(registry, cache) as client:
...     run_id = client.start_flow(orders, "order-42", OrderInput("42"))
...     result = client.wait_for_flow("order-42").single_output(OrderResult)
Client( registry: Registry, blob_cache: BlobCache, options: ClientOptions | None = None)

Construct a Client and its lazy plaintext gRPC channel.

Arguments:
  • registry: The Flow Registry used for call validation and codecs.
  • blob_cache: An open cache used to hydrate large response values.
  • options: Service address and bearer credential; None uses defaults and sends no authorization metadata.
registry
blob_cache
options
def start_flow( self, flow: Flow[~InputT], flow_id: str, input: ~InputT, options: StartFlowOptions = StartFlowOptions(timeout=None, timeout_policy=<FlowTimeoutPolicy.DEFAULT: 'default'>, timeout_handler_options=None, start_delay=None, id_reuse_policy=<IdReusePolicy.DEFAULT: 'default'>, retry_policy=None, _attribute_initializations=(), config_override=None, ignore_already_started=False, request_id=None)) -> str:

Start a Flow and return its server-assigned run ID.

The call returns after Dex accepts the Flow, not after completion. flow must be the exact instance registered with this Client, and input must match its starting Step annotation. Options control timeout, reuse, retries, configuration, idempotency, and initial Attributes.

The first request negotiates Server protocol 4 and caches compatibility. Construction performs no network I/O. An incompatible Server fails this and every later request; transient negotiation failures can be retried.

Arguments:
  • flow: The registered Flow instance to start.
  • flow_id: A non-empty application ID stable across runs.
  • input: The starting Step input, or None when no starting Step exists.
  • options: Settings for the new Flow execution.
Returns:

The server-assigned run ID.

Raises:
  • FlowAlreadyStartedError: If reuse policy rejects an existing Flow ID.
  • FlowDefinitionError: If flow is not registered.
  • ValueMappingError: If input or an initial Attribute cannot be encoded, including a numeric or Boolean value for a hard-delete Attribute.
  • ValueError: If an initial Attribute uses a different hard_delete_on_null setting than the Flow's registered definition of that name.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
  • DexServiceError: If FlowService rejects or cannot perform the request.
def invoke_rpc( self, rpc_method: Callable[..., Any], flow_id: str, input: object = None, *, run_id: str = '', options: RPCInvokeOptions = RPCInvokeOptions(lock_attribute_map_instances=(), load_attribute_map_instances=(), load_channel_map_instances=())) -> Any:

Synchronously invoke a registered RPC on a Flow execution.

Pass the bound method from the same registered Flow instance. RPC timeout and Attribute locks come from its @rpc decorator. The call blocks until the handler returns, fails, or times out. A non-transactional RPC without Attribute locks starts from a backend query. If the handler returns no durable effects, a retained terminal execution can serve that query. Locks, transactional execution, returned effects, or server policy can require an active execution.

Arguments:
  • rpc_method: A bound method decorated with @rpc.
  • flow_id: The non-empty target Flow ID.
  • input: The annotated RPC input, or None for an input-free RPC.
  • run_id: Optional exact run. With "", the server resolves the current execution.
  • options: Additive runtime AttributeMap locks and exact map-instance loads.
Returns:

The decoded RPCResult.output, or None for a no-output RPC.

Raises:
  • FlowDefinitionError: If the method is not registered.
  • RpcLockConflictError: If requested Attribute locks cannot be acquired.
  • WorkerInvocationError: If the application handler fails.
  • ValueMappingError: If input or output mapping fails.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or the selected path cannot use its closed execution.
  • DexServiceError: If the service call otherwise fails.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def write_stream( self, flow_id: str, stream: Stream[~ValueT], source: str, value: ~ValueT) -> None:

Append one typed best-effort Stream message with source metadata.

Arguments:
  • flow_id: Logical Flow instance ID; the Flow need not exist or be active.
  • stream: Exact Stream object registered in one Flow schema.
  • source: Non-empty informational source. Repeated values still append.
  • value: Typed message to append.
Raises:
  • ValueError: If the Flow ID or source is empty.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If the message cannot be encoded.
  • DexServiceError: If FlowService cannot append the message.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def read_stream( self, flow_id: str, stream: Stream[~ValueT], resume_token: str = '', timeout: datetime.timedelta | None = None) -> StreamMessage[~ValueT]:

Block for the next retained Stream message after a resume token.

Arguments:
  • flow_id: Logical Flow instance ID used as the Stream instance key.
  • stream: Exact Stream object registered in one Flow schema.
  • resume_token: Previous message token, or empty for the retained head.
  • timeout: Optional non-negative server-side long-poll duration.
Returns:

The decoded message, next resume token, creation time, and source.

Raises:
  • LongPollTimeoutError: If no message arrives before the server wait expires.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If the retained message cannot be decoded.
  • DexServiceError: If FlowService cannot perform the read.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def list_stream_messages( self, flow_id: str, stream: Stream[~ValueT], page_size: int, before_page_token: str = '') -> StreamMessagesPage[~ValueT]:

Return one newest-first page of retained Stream messages without waiting.

An empty token starts at the retained tail. Pass the returned page token unchanged to read older messages. Trimming may remove messages between pages.

Arguments:
  • flow_id: Logical Flow instance ID used as the Stream instance key.
  • stream: Exact Stream object registered in one Flow schema.
  • page_size: Positive maximum message count within the server-configured limit.
  • before_page_token: Opaque token from the preceding page, or "" first.
Returns:

Decoded messages in newest-first order and the next-page token.

Raises:
  • ValueError: If page_size is not positive.
  • FlowDefinitionError: If the Stream is not registered.
  • ValueMappingError: If a retained message cannot be decoded.
  • DexServiceError: If FlowService cannot list the Stream.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def wait_for_flow( self, flow_id: str, timeout: datetime.timedelta | None = None) -> FlowResult:

Block until a Flow closes and return its terminal result.

timeout is the server-side long-poll duration. A long-poll timeout is retryable by calling this method again. Parallel completion order is not deterministic; select results by Step identity rather than position.

Arguments:
  • flow_id: The non-empty target Flow ID.
  • timeout: Optional non-negative server-side wait duration.
Returns:

The terminal status, failure detail, and output-bearing completions.

Raises:
  • LongPollTimeoutError: If the Flow remains open when the wait expires.
  • FlowNotFoundError: If flow_id does not exist.
  • ValueMappingError: If Dex returns a malformed or incompatible hydrated value.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def stop_flow( self, flow_id: str, options: StopFlowOptions = StopFlowOptions(type=<StopType.CANCEL: 'cancel'>, reason=None)) -> None:

Request cancellation, termination, or failure of an active Flow.

The call returns after Dex accepts the request and does not wait for terminal status.

Arguments:
  • flow_id: The non-empty target Flow ID.
  • options: Stop mode and optional recorded reason.
Raises:
  • FlowNotActiveOrNotFoundError: If no active run exists.
  • DexServiceError: If FlowService rejects or cannot perform the request.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def describe_flow(self, flow_id: str) -> FlowInfo:

Return summary metadata for the current or latest Flow run.

Arguments:
  • flow_id: The non-empty Flow ID to describe.
Returns:

Flow ID, run ID, Flow type, status, and UTC start time.

Raises:
  • FlowNotFoundError: If flow_id does not exist.
  • DexServiceError: If FlowService cannot return the summary.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def get_attributes( self, flow_id: str, *reads: Union[Attribute[Any], AttributeMapInstance[Any]]) -> AttributeValues:

Read committed Attribute values from the current or latest Flow run.

Use get_attributes to read Flow state from outside the Flow without defining an RPC. Pass one or more reads. A read is a singleton Attribute or one AttributeMap instance selected with AttributeMap.instance(key). To read one Attribute, pass a single read. The Client sends every key in one FlowService request, so all values come from the same committed state. No Worker runs.

The call targets the current run, or the latest run when no run is open. A closed run stays readable until Dex purges it. The call sees only committed state. Unlike reading inside an RPC handler, it never sees writes that a running Step or RPC has staged but not yet committed.

Each read must use a definition that some Flow in this Client's Registry declares in its PersistenceSchema. The definition matches when it is equal to the registered one: same class (Attribute or AttributeMap), name, value_type, index, sync_to_attribute_store, and hard_delete_on_null. The definition may come from any registered Flow. The server does not check that the target Flow's type declares it, so a same-named Attribute of another type in the target Flow fails to decode.

Attributes defined with hard_delete_on_null=True are readable; the server resolves their stored values. The Client resolves other large values through LoadBlobs and the Client's BlobCache, and then decodes every value. On any error the call returns nothing, so no partial result is visible. Every error message names the Flow ID. An error about one read or value also names its Attribute name or physical key.

Example::

ORDER_STATUS = Attribute("order-status", str)
NOTES = AttributeMap("notes", Note)

try:
    values = client.get_attributes(
        "order-42", ORDER_STATUS, NOTES.instance("n1")
    )
except FlowNotFoundError:
    return None
status = values.get(ORDER_STATUS, "new")
if NOTES.instance("n1") in values:
    note = values[NOTES.instance("n1")]
Arguments:
  • flow_id: The non-empty Flow ID.
  • *reads: At least one Attribute or AttributeMapInstance. Two reads of the same Attribute or AttributeMap instance are rejected.
Returns:

The set values, keyed by the reads passed to this call. An Attribute or AttributeMap instance without a stored value is absent from the result.

Raises:
  • ValueError: If flow_id is empty, no read is given, two reads select the same Attribute or AttributeMap instance, or the response has an unexpected key, a duplicate key, or an entry without a value.
  • TypeError: If a read is not an Attribute or AttributeMapInstance, such as an AttributeMap without an instance.
  • FlowDefinitionError: If no Flow in the Registry declares a matching definition, including a same-named definition with another value type.
  • FlowNotFoundError: If the Flow does not exist or its latest run is being purged.
  • ValueMappingError: If a returned value cannot be resolved or decoded as its value type. This includes an empty or non-finite value, an unsupported object encoding, and a large value that LoadBlobs omits.
  • DexServiceError: If FlowService or LoadBlobs otherwise fails. A LoadBlobs failure for a missing Flow raises FlowNotFoundError.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def search_flows( self, query: str, page_size: int, next_page_token: str = '', *, include_continued_as_new: bool = False) -> SearchFlowsPage:

Return one page of Flow runs matching a visibility query.

Arguments:
  • query: A Dex visibility query; an empty string matches server defaults.
  • page_size: The non-negative maximum result count requested.
  • next_page_token: Opaque token from the preceding page, or "" first.
  • include_continued_as_new: Include earlier runs without overriding query filters.
Returns:

Server-ordered entries with hydrated indexed Attributes and a next token.

Raises:
  • ValueError: If page_size is negative.
  • ValueMappingError: If an indexed Attribute cannot be hydrated.
  • DexServiceError: If FlowService cannot execute the query.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def time_travel(self, flow_id: str, options: TimeTravelOptions) -> str:

Create a new run from a selected point in existing Flow history.

Arguments:
  • flow_id: The non-empty Flow ID whose history supplies the new run.
  • options: Time travel selector, reason, and replay controls.
Returns:

The new server-assigned run ID; the Flow ID is unchanged.

Raises:
  • FlowNotFoundError: If flow_id does not exist.
  • ValueError: If time travel selector fields are invalid.
  • DexServiceError: If FlowService cannot create the new run.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def skip_timer( self, flow_id: str, step_execution_id: StepExecutionId, timer_id: TimerId) -> None:

Make one waiting Timer condition immediately ready.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • step_execution_id: The Step type and positive execution number.
  • timer_id: Exactly one Timer condition ID or zero-based index.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • ValueError: If an identifier is invalid.
  • DexServiceError: If FlowService cannot skip the Timer.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def wait_for_step_completion( self, flow_id: str, step_execution_id: StepExecutionId, options: WaitForStepCompletionOptions) -> None:

Block until one Step execution completes or its request timeout expires.

The server derives a stable Request ID from the Step execution when none is supplied. Request timeout covers transparent transport reattachments. Internal handler timeout controls Update generation rollover without ending this request.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • step_execution_id: The Step type and positive execution number.
  • options: The optional Request ID, request timeout, and internal handler timeout controls.
Raises:
  • ValueError: If the Request ID or request timeout is invalid.
  • RequestTimeoutError: If a positive request timeout expires first.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closes first.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def wait_for_attribute_match( self, flow_id: str, attribute: Union[Attribute[Any], AttributeMap[Any]], *args: object, **kwargs: object) -> Any:

Wait for a singleton Attribute or AttributeMap instance to match.

Singleton form is wait_for_attribute_match(flow_id, attribute, match, options); map form adds instance before match.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • attribute: The registered Attribute or AttributeMap to observe.
  • *args: Positional match and options, optionally preceded by a map instance.
  • **kwargs: The same arguments supplied by name.
Returns:

The current value that satisfied the match.

Raises:
  • TypeError: If arguments do not match the Attribute definition.
  • ValueError: If an identifier, option, or expected value is invalid. Also raised before transport when the Attribute uses hard_delete_on_null.
  • RequestTimeoutError: If a positive request timeout expires first.
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closes first.
  • DexServiceError: If FlowService cannot perform the wait.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def update_flow_config(self, flow_id: str, config: FlowConfig) -> None:

Update specified mutable configuration fields for an active Flow.

The update affects later decisions and does not recall work already dispatched. Unspecified fields and the existing Worker target remain unchanged. An empty FlowConfig() leaves all configuration unchanged.

Arguments:
  • flow_id: The non-empty active Flow ID.
  • config: The new optional configuration fields.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • ValueError: If a configuration value is invalid.
  • DexServiceError: If FlowService cannot apply the update.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def trigger_continue_as_new(self, flow_id: str) -> None:

Ask an active Flow to roll history into a new run.

The request returns after acceptance; the Flow ID remains stable and the run ID changes when rollover occurs.

Arguments:
  • flow_id: The non-empty active Flow ID.
Raises:
  • FlowNotActiveOrNotFoundError: If the Flow is missing or closed.
  • DexServiceError: If FlowService cannot accept the request.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def health_check(self) -> bool:

Call the FlowService health endpoint.

Returns:

True when the service returns successfully.

Raises:
  • DexServiceError: If the service is unhealthy or unreachable.
  • RuntimeError: If the Server reports a protocol range without protocol 4.
def close(self) -> None:

Close the owned gRPC channel.

close is idempotent. Other Client methods must not be called afterward. The Registry and BlobCache remain owned by the caller.

@dataclass(frozen=True)
class ClientOptions:

Configure FlowService connectivity and bearer credentials.

Attributes:
  • server_address: The plaintext gRPC target for Dex. Defaults to "localhost:8801" and must not include an HTTP scheme.
  • credential: Supplies the current bearer token for every FlowService call. None sends no authorization metadata. A token error fails the call before contacting Dex. AsyncClient accepts an awaitable token and bounds its lookup by the RPC timeout when set. Synchronous token lookup must return promptly. Protect the plaintext network path.
ClientOptions( server_address: str = 'localhost:8801', credential: Credential | None = None)
server_address: str = 'localhost:8801'
credential: Credential | None = None
class Credential(typing.Protocol):

Supply a bearer token for each Client or Worker service call.

token may be called concurrently. AsyncClient and AsyncWorker accept an awaitable token; synchronous callers require a string. Synchronous tokens run on the calling thread for sync APIs and in an SDK-owned, four-thread executor for async APIs. Each Client or Worker creates its executor lazily and closes it without waiting for a blocked token call. A blocked token keeps one executor thread busy until it returns. An unclosed Client or Worker may retain up to four idle credential threads. Declare token with async def to invoke it on the event loop. A synchronous token method may return a coroutine, which is awaited on the event loop. A loop-bound Future from another loop raises TypeError. Async calls bound token lookup by the RPC deadline when one is set. Pull stream opens use a 30-second lookup budget. Sync calls do not bound token lookup separately, so synchronous implementations must return promptly. Client calls raise a token exception unchanged, before contacting Dex. Worker startup may wrap it with startup context. The token travels over Dex's plaintext gRPC connection; protect the network path.

Example:
>>> class FixedCredential:
...     def token(self) -> str:
...         return "my-token"
>>> credential = FixedCredential()
>>> options = ClientOptions(credential=credential)
Credential(*args, **kwargs)
def token(self) -> Union[str, Awaitable[str]]:

Return the current bearer token, or raise when unavailable.

Returns:

Token without the Bearer prefix, or an awaitable for async callers.

Raises:
  • Exception: Any failure prevents the service call from reaching Dex.
class Codec(typing.Protocol[~ValueT]):

Define bidirectional conversion between a Python type and Dex Values.

Custom codecs must be deterministic: equivalent inputs should produce the same Value so durable replays observe identical data.

Codec(*args, **kwargs)
type_name: str

Return the application-facing type name used in error messages.

Returns:

A stable, non-empty type name.

wire_kind: WireKind

Return the single wire kind emitted and accepted by this codec.

Returns:

The codec's protocol representation.

def encode(self, value: ~ValueT) -> Value:

Encode one typed Python value.

Arguments:
  • value: The application value to encode.
Returns:

A Value whose kind equals wire_kind.

Raises:
  • TypeError: If value is incompatible with the codec.
  • ValueError: If the value is not representable by Dex.
def decode(self, value: Value) -> ~ValueT:

Decode one protocol Value into the application type.

Arguments:
  • value: The validated Value to decode.
Returns:

The decoded application value.

Raises:
  • TypeError: If the wire kind or payload is incompatible.
  • ValueError: If the payload is malformed.
class CodecRegistry:

Resolve codecs for type hints used by registered SDK definitions.

Explicit registrations take precedence over built-ins. The default registry supports primitive types, None, datetime, dataclasses, enums, and typed list, tuple, mapping, and sequence containers. Parameterized generic dataclasses substitute their type arguments recursively. Optional types and unions use JSON, including when their alternatives have primitive codecs.

Unions select an alternative by JSON kind: null, boolean, number, string, array, or object. Boolean values never select an integer alternative. A union passed directly to resolve() needs alternatives with disjoint kinds. For example, Receipt | None and int | str work, but int | float, str | datetime, and unions of two dataclasses are ambiguous and raise TypeError during resolution. A custom codec can implement a discriminator. A Literal alternative, such as Literal["draft", "sent"] | None, uses the JSON kinds of its members. Decoding raises ValueError for a value outside the Literal. Members match by exact type, so JSON true never matches Literal[1].

Unions inside dataclass fields and containers keep the decoding of earlier releases and never raise during resolution. Such a union decodes a value into a dataclass, Enum, or datetime alternative, or a container of those, when the value's JSON kind matches only that alternative. Otherwise it returns the JSON value unchanged, without a type check. A nested Literal also returns the JSON value unchanged. Fields typed by a type variable and Mapping or Sequence types follow the same rule: Result[Receipt] decodes a Receipt, but Result[int] returns the JSON value of its type-variable field unchanged. A float accepts a JSON integer and keeps it as an int.

Dataclass fields use their Python names unless field(metadata={"dex_json_name": "callId"}) specifies a non-empty wire name. Names must be unique within the dataclass. Unknown input keys are ignored. Absent keys use the declared default or default factory; a missing required field raises ValueError. Explicit null for a defaulted field still decodes as null. It raises TypeError when the field type checks values, such as int, a dataclass, or datetime.

dex_json_omit_default=True in field metadata omits values equal to the field's default. It requires a default or default factory. A factory is called again for comparison on every encode, so it must be deterministic and free of side effects. Without this opt-in, defaults and nulls are included in encoded JSON.

Datetime fields decode ISO 8601 strings, including Go RFC3339 strings with Z or numeric timezone offsets. Python retains microsecond precision; additional fractional digits are truncated on decode. Encoding retains datetime.isoformat() by default. Field metadata dex_json_format="rfc3339" instead emits Go-compatible timestamps with Z for UTC and no trailing fractional zeroes. This opt-in requires a datetime or optional datetime field and an aware value whose timezone offset is a whole number of minutes.

Examples:
>>> registry = CodecRegistry({Money: money_codec})
>>> registry.resolve(Money) is money_codec
True

A generic dataclass needs no explicit codec registration::

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Generic, TypeVar
from dex import CodecRegistry

ResultT = TypeVar("ResultT")

@dataclass
class Receipt:
    call_id: str = field(metadata={"dex_json_name": "callId"})
    observed_at: datetime = field(metadata={
        "dex_json_name": "observedAt", "dex_json_format": "rfc3339"
    })

@dataclass
class Result(Generic[ResultT]):
    value: ResultT
    receipt: Receipt | None = None
    failure: str | None = field(default=None, metadata={
        "dex_json_omit_default": True
    })

codec = CodecRegistry().resolve(Result[int])
result = Result(42, Receipt("call-1", datetime.now(timezone.utc)))
assert codec.decode(codec.encode(result)) == result
CodecRegistry(codecs: Mapping[object, Codec[typing.Any]] | None = None)

Create a registry with optional custom type-hint mappings.

Arguments:
  • codecs: Custom codecs keyed by the exact type hint used in definitions. The mapping is copied and may be mutated after construction.
def resolve(self, type_hint: object) -> Codec[typing.Any]:

Return the codec for an exact type hint.

Arguments:
  • type_hint: A runtime type or supported parameterized type hint.
Returns:

The registered, built-in, or automatically derived codec.

Raises:
  • TypeError: If no codec can represent type_hint, type_hint is a union with overlapping or unsupported alternatives, or dataclass JSON metadata is invalid.
@dataclass(frozen=True)
class ConditionCombination:

Group Conditions that must become ready together.

Pass combinations to Wait.any_combination_of() when several alternative groups are acceptable.

Attributes:
  • conditions: The ordered Conditions in this all-of group.
ConditionCombination(conditions: tuple[dex.condition.Condition, ...])
conditions: tuple[dex.condition.Condition, ...]
@staticmethod
def of( *conditions: dex.condition.Condition) -> ConditionCombination:

Create a combination from one or more Conditions.

Arguments:
  • *conditions: Conditions that must all become ready together.
Returns:

A combination with Conditions in argument order.

class Context(typing.Protocol):

Expose execution metadata and decision-local Step operations.

Dex supplies a Context to every wait_for and execute call. Do not retain it after the handler returns. Attribute and Channel definition methods provide the typed persistence API; the methods below expose execution diagnostics and per-attempt local state.

Context(*args, **kwargs)
flow_id: str

Return the stable Flow ID shared by all runs.

Returns:

The application-assigned Flow ID.

run_id: str

Return the current server-assigned run ID.

Returns:

The run ID, which changes after time travel or continue-as-new.

flow_started_at: datetime.datetime

Return when the current Flow run started.

The server supplies the same value across Steps and handler retries in this run. It is available in wait_for, execute, Flow timeout handlers, and RPC handlers. This is the run start time, rather than the start time of the current Step or attempt.

Returns:

A timezone-aware UTC datetime with one-second precision.

Examples:

Measure the current run's age inside a handler::

from datetime import datetime, timezone

elapsed = datetime.now(timezone.utc) - context.flow_started_at
elapsed_seconds = elapsed.total_seconds()
step_execution_id: str

Return the current Step execution's protocol identifier.

Returns:

The Step type and execution number encoded by Dex.

from_step_execution_id: str

Return the predecessor Step execution identifier.

Returns:

The originating identifier, or an empty string at Flow start.

recovery_error: RecoveryErrorInfo | None

Return the final failure that selected this recovery path.

Returns:

Recovery information, or None outside a recovery path.

first_attempt_at: datetime.datetime

Return when the first attempt of this handler invocation started.

The value is stable across retries of the same invocation, so a handler can derive a deadline that a retry does not reset.

Returns:

A timezone-aware UTC time with one-second precision.

deadline: datetime.datetime | None

Return the current Step method attempt's absolute deadline.

RetryPolicy total duration does not shorten a running attempt. Use this inside wait_for, execute, or a Flow timeout handler to budget external work; each retry receives its own effective deadline.

Returns:

A timezone-aware UTC datetime with microsecond precision, or None for RPC handlers and tasks without a server-supplied deadline.

Examples:

Compute the remaining attempt budget in seconds inside a handler::

from datetime import datetime, timezone

deadline = context.deadline
if deadline is not None:
    remaining_seconds = (deadline - datetime.now(timezone.utc)).total_seconds()
attempt: int

Return the one-based handler retry attempt.

Returns:

1 for the initial attempt and a larger value after retries.

def has_last_heartbeat_value(self) -> bool:

Report whether the previous regular attempt persisted heartbeat details.

Returns:

True when a heartbeat Value is present. A persisted Python None still counts as present.

def get_last_heartbeat_value(self, value_type: type[~ValueT]) -> Optional[~ValueT]:

Decode the previous regular attempt's heartbeat checkpoint.

Use has_last_heartbeat_value() to distinguish an absent checkpoint from a persisted Python None.

Arguments:
  • value_type: The expected Python type and codec selection.
Returns:

The decoded checkpoint, or None when no Value is present.

Raises:
  • ValueMappingError: If the Value cannot be decoded as value_type.
def has_timer_fired(self, index: int | None = None) -> bool:

Report whether a Timer made the current Wait ready.

Arguments:
  • index: Optional zero-based Timer index; None checks any Timer.
Returns:

True when the selected Timer fired for this execution.

def wait_for_method_failed(self) -> bool:

Report whether wait_for failed before this execute call.

Returns:

True when the Step's failure policy proceeded after a wait error.

def is_cancellation_requested(self) -> bool:

Report whether Dex canceled the active handler call.

Async handlers are normally canceled by their task. Synchronous handlers cannot be forcefully interrupted by Python; long CPU-bound work may check this at natural boundaries. External effects still require idempotency or compensation.

Returns:

True after the Worker gRPC call has been canceled.

def set_step_execution_local(self, key: str, value: object) -> None:

Store process-local data for this Step execution.

Local data is not persisted, replicated, or available after worker restart; use Attributes for durable state.

Arguments:
  • key: A non-empty key scoped to the Step execution.
  • value: The in-memory object to retain by reference.
def get_step_execution_local(self, key: str, value_type: type[~ValueT]) -> Optional[~ValueT]:

Return typed process-local data for this Step execution.

Arguments:
Returns:

The stored object, or None when the key is absent.

Raises:
  • TypeError: If the stored object is not an instance of value_type.
def record_event(self, name: str, value: object) -> None:

Stage an application event in the current handler result.

Arguments:
  • name: The non-empty event name used for diagnostics.
  • value: A codec-supported event payload.
class DexServiceError(builtins.RuntimeError):

Represent a structured error returned by a Dex FlowService operation.

Attributes:
  • code: The outer gRPC status code.
  • sub_status: The stable Dex-specific error classification.
  • detail: The human-readable service detail.
  • operation: The Client operation that failed.
  • flow_id: The targeted Flow ID, or None for service-wide calls.
DexServiceError( code: grpc.StatusCode, sub_status: ErrorSubStatus, detail: str, operation: str, flow_id: str | None)

Create a structured FlowService error.

Arguments:
  • code: The outer gRPC status code.
  • sub_status: The Dex-specific classification.
  • detail: Human-readable error detail.
  • operation: The Client operation name.
  • flow_id: The targeted Flow ID, if any.
code
sub_status
detail
operation
flow_id
class ErrorSubStatus(enum.Enum):

Classify a FlowService error more precisely than its gRPC status.

Attributes:
  • UNCATEGORIZED: Dex returned no specific classification.
  • FLOW_ALREADY_STARTED: The requested Flow ID conflicts with an existing run.
  • FLOW_NOT_EXISTS: No Flow with the requested ID exists.
  • WORKER_API_ERROR: An application Worker rejected or failed an invocation.
  • LONG_POLL_TIMEOUT: A wait ended without observing its condition.
  • CHANNEL_MESSAGE_NOT_FOUND: A pending Channel message ID no longer exists.
  • REQUEST_TIMEOUT: A caller-defined request budget expired.
  • FLOW_CLOSING: Dex rejected an update-class request during Flow close cleanup.
UNCATEGORIZED = <ErrorSubStatus.UNCATEGORIZED: 'uncategorized'>
FLOW_ALREADY_STARTED = <ErrorSubStatus.FLOW_ALREADY_STARTED: 'flow_already_started'>
FLOW_NOT_EXISTS = <ErrorSubStatus.FLOW_NOT_EXISTS: 'flow_not_exists'>
WORKER_API_ERROR = <ErrorSubStatus.WORKER_API_ERROR: 'worker_api_error'>
LONG_POLL_TIMEOUT = <ErrorSubStatus.LONG_POLL_TIMEOUT: 'long_poll_timeout'>
CHANNEL_MESSAGE_NOT_FOUND = <ErrorSubStatus.CHANNEL_MESSAGE_NOT_FOUND: 'channel_message_not_found'>
REQUEST_TIMEOUT = <ErrorSubStatus.REQUEST_TIMEOUT: 'request_timeout'>
FLOW_CLOSING = <ErrorSubStatus.FLOW_CLOSING: 'flow_closing'>
class Flow(typing.Generic[~StartT], abc.ABC):

Define a durable application Flow and its registered API surface.

Subclasses return their Steps and persistence schema and may expose methods decorated with rpc(). Construct one Flow instance and pass it to Registry; Client calls must use that same registered instance.

def get_flow_type(self) -> str:

Return the protocol Flow type used by Client and Dex.

Override it to decouple the type from the Python class name.

Returns:

A non-empty Flow type unique within the Registry.

def get_steps(self) -> StepList[~StartT]:

Return the Step definitions for this Flow.

Returns:

A StepList with zero or one starting Step and all reachable Steps.

def get_persistence_schema(self) -> PersistenceSchema:

Return the Attributes and Channels owned by this Flow.

Returns:

The persistence schema; the default is empty.

def handle_timeout(self, context: Context) -> StepDecision:

Handle expiration of this Flow's durable soft-timeout timer.

Override this method to make a positive timeout default to HANDLER. Async Flow implementations may use async def with the same annotations. The returned StepDecision uses ordinary Execute validation and may transition, dead-end, complete, fail, or request graceful completion.

Arguments:
  • context: The timeout-handler invocation Context; do not retain it.
Returns:

The decision applied after the timeout timer completes or is skipped.

Raises:
  • NotImplementedError: If called without an application override.
class FileCredential:

Read a bearer token from a file and pick up rotations automatically.

The file is read during construction. Its mtime and size are checked at most once per second, and changed contents are reloaded before the next call. Surrounding whitespace is removed. A file that becomes empty or unreadable fails subsequent calls until restored. Concurrent calls share a lock. Async callers check the file in an executor, outside the event loop.

Example:
>>> credential = FileCredential("/var/run/secrets/dex/token")
>>> client = Client(registry, cache, ClientOptions(credential=credential))
Arguments:
  • path: Path to the token file.
Raises:
  • OSError: The file cannot be read during construction.
  • ValueError: The file has no token after trimming whitespace.
  • UnicodeError: The file does not contain valid UTF-8 text.
FileCredential(path: str | os.PathLike[str])

Read the initial token from path and fail if it is unavailable or empty.

Arguments:
  • path: File containing the bearer token.
Raises:
  • OSError: The file cannot be read.
  • ValueError: The file has no token after trimming whitespace.
  • UnicodeError: The file does not contain valid UTF-8 text.
def token(self) -> str:

Return the current trimmed token, checking for file changes once per second.

Returns:

The token without the Bearer prefix.

Raises:
  • OSError: The file is no longer readable.
  • ValueError: The file is empty after trimming whitespace.
  • UnicodeError: The file does not contain valid UTF-8 text.
class FlowAlreadyStartedError(dex.DexServiceError):

Indicate that start_flow conflicts with an existing Flow ID.

class FlowClosingError(dex.DexServiceError):

Report a request rejected while Dex closes a Flow.

Re-read the retained terminal Flow state before deciding what to do next. Catch this error directly instead of comparing the service detail::

from dex import FlowClosingError, WaitForStepCompletionOptions

try:
    await client.wait_for_step_completion(
        flow_id, step_execution_id, WaitForStepCompletionOptions()
    )
except FlowClosingError:
    terminal_result = await client.wait_for_flow(flow_id)
@dataclass(frozen=True)
class FlowConfig:

Override server behavior for one Flow execution.

At start time, every None field uses the registered or server default. Client configuration updates preserve existing values for None fields.

Attribute Store projection is asynchronous. Failures never roll back Flow Attributes, and already queued projections retain their original target.

Attributes:
  • active_step_search_mode: Optional active-Step visibility indexing policy.
  • continue_as_new_threshold: Optional positive history-event threshold that requests continue-as-new.
  • continue_as_new_page_size_bytes: Optional positive history page-size budget in bytes used by continue-as-new decisions.
  • step_durability: Optional default durability for Step handlers.
  • attribute_store_names: Optional Server-configured Attribute Store names. None leaves the field absent; an empty list disables future projections.
FlowConfig( active_step_search_mode: ActiveStepSearchMode | None = None, continue_as_new_threshold: int | None = None, continue_as_new_page_size_bytes: int | None = None, step_durability: StepDurability | None = None, attribute_store_names: list[str] | None = None)
active_step_search_mode: ActiveStepSearchMode | None = None
continue_as_new_threshold: int | None = None
continue_as_new_page_size_bytes: int | None = None
step_durability: StepDurability | None = None
attribute_store_names: list[str] | None = None
class FlowDefinitionError(builtins.RuntimeError):

Indicate an invalid Flow definition or a definition the Registry lacks.

Registry construction raises it for an invalid Flow definition. Client calls raise it before contacting Dex when an argument is not registered or does not match the registered definition. For example, get_attributes raises it for an Attribute whose value type differs from the one a Flow declares.

class FlowErrorType(enum.Enum):

Identify the terminal failure category reported for a Flow.

Attributes:
  • STEP_DECISION_FAILED: A Step decision could not be applied.
  • CLIENT_API_FAILED: A Client-originated operation failed the Flow.
  • WORKER_API_FAILED: A Worker handler or dispatch failed.
  • INVALID_USER_FLOW_CODE: Application Flow definitions or results were invalid.
  • INTERNAL: Dex encountered an internal invariant or infrastructure failure.
  • FLOW_TIMEOUT: A Dex soft Flow timeout expired under the fail policy.
STEP_DECISION_FAILED = <FlowErrorType.STEP_DECISION_FAILED: 'step_decision_failed'>
CLIENT_API_FAILED = <FlowErrorType.CLIENT_API_FAILED: 'client_api_failed'>
WORKER_API_FAILED = <FlowErrorType.WORKER_API_FAILED: 'worker_api_failed'>
INVALID_USER_FLOW_CODE = <FlowErrorType.INVALID_USER_FLOW_CODE: 'invalid_user_flow_code'>
INTERNAL = <FlowErrorType.INTERNAL: 'internal'>
FLOW_TIMEOUT = <FlowErrorType.FLOW_TIMEOUT: 'flow_timeout'>
@dataclass(frozen=True)
class FlowInfo:

Summarize the current or latest run of one Flow ID.

Attributes:
  • flow_id: The stable application-assigned Flow ID.
  • run_id: The server-assigned run ID.
  • flow_type: The registered Flow type name.
  • status: The run's current lifecycle status.
  • started_at: The timezone-aware server start timestamp.
FlowInfo( flow_id: str, run_id: str, flow_type: str, status: FlowStatus, started_at: datetime.datetime)
flow_id: str
run_id: str
flow_type: str
status: FlowStatus
started_at: datetime.datetime
class FlowResult:

Describe an observed Flow status and its output-bearing completions.

Client results are terminal. A SubFlow result can be a running snapshot when another AnyOf Condition won; that status is not a live backend query.

Attributes:
  • status: The observed lifecycle state.
  • error_type: Dex failure category when available.
  • error_message: Server failure detail when available.
  • completions: An immutable sequence of Step completions.
FlowResult( status: FlowStatus, completions: Sequence[StepCompletion], error_type: FlowErrorType | None = None, error_message: str | None = None)

Create an immutable Flow result snapshot.

Arguments:
  • status: The observed Flow status.
  • completions: Output-bearing Step completions in server collection order.
  • error_type: Optional Dex failure category.
  • error_message: Optional server failure detail.
status: FlowStatus

Return the observed Flow lifecycle state.

Returns:

The status captured by this result.

error_type: FlowErrorType | None

Return the Dex failure category, if one was reported.

Returns:

The failure category, or None when Dex reported none.

error_message: str | None

Return the server failure detail, if one was reported.

Returns:

The failure detail, or None when Dex reported none.

is_terminal: bool

Return whether this observed run can no longer execute.

Returns:

True for a terminal snapshot; otherwise False.

completions: tuple[StepCompletion, ...]

Return completions in server collection order.

Returns:

An immutable tuple of output-bearing Step completions.

def single_output(self, output_type: type[~OutputT]) -> ~OutputT:

Decode the output when exactly one completion exists.

Arguments:
  • output_type: The concrete Python type expected by the caller.
Returns:

The only decoded Step output.

Raises:
  • ValueError: If the result is nonterminal or has zero or multiple outputs.
  • ValueMappingError: If the output cannot be decoded as output_type.
class FlowNotActiveOrNotFoundError(dex.DexServiceError):

Report a missing Flow or an operation that cannot use a closed Flow.

RPC calls use this error for missing targets, including query-only reads. A query-only RPC can read a retained terminal execution when it has no locks, transactionality, durable effects, or Server-forced Update routing. At that confirmed read boundary, handle this error as not found without adding a lifecycle probe. Preserve other service and Worker failures.

Mutations and active-only RPCs also use this error for closed targets. The error does not distinguish a missing Flow from a closed Flow or prove that the requested action succeeded. The inherited service metadata and chained transport cause remain available. For example::

try:
    client.invoke_rpc(orders.update_order, order_id, update)
except FlowNotActiveOrNotFoundError:
    raise OrderUnavailableError(order_id)
class FlowNotFoundError(dex.DexServiceError):

Indicate that an operation targeted a Flow ID that does not exist.

class FlowStatus(enum.Enum):

Describe the lifecycle state of a Flow run.

Attributes:
  • RUNNING: The Flow can still accept work and state changes.
  • COMPLETED: The Flow completed successfully.
  • FAILED: The Flow ended because application or worker processing failed.
  • CANCELED: The Flow cooperatively accepted a cancellation request.
  • TERMINATED: The Flow was forcibly terminated.
  • SERVER_SIDE_TIMEOUT_INTERNAL_ONLY: Reserved backend hard-timeout status.
  • CONTINUED_AS_NEW: This run rolled forward into a new run.
RUNNING = <FlowStatus.RUNNING: 'running'>
COMPLETED = <FlowStatus.COMPLETED: 'completed'>
FAILED = <FlowStatus.FAILED: 'failed'>
CANCELED = <FlowStatus.CANCELED: 'canceled'>
TERMINATED = <FlowStatus.TERMINATED: 'terminated'>
SERVER_SIDE_TIMEOUT_INTERNAL_ONLY = <FlowStatus.SERVER_SIDE_TIMEOUT_INTERNAL_ONLY: 'server_side_timeout_internal_only'>
CONTINUED_AS_NEW = <FlowStatus.CONTINUED_AS_NEW: 'continued_as_new'>
@dataclass(frozen=True)
class FlowTimeoutHandlerOptions:

Configure timeout-handler execution and selective state loading.

Ordinary Attributes and Channel size metadata load automatically. AttributeMap values and pending Channel messages require an explicit selection. One logical handler execution may contain multiple retry attempts.

Attributes:
  • method_timeout: Maximum duration of one handler attempt.
  • heartbeat_timeout: Progress deadline for one regular attempt.
  • retry: Optional retry policy; None uses server defaults.
  • durability: Durability for successful handler side effects.
  • lock_attributes: Attribute locks held during the handler invocation.
  • load_attribute_maps: AttributeMaps whose current instances load.
  • load_attribute_map_instances: Exact AttributeMap instances to load.
  • load_channels: Channels whose pending messages load.
  • load_channel_maps: ChannelMaps whose current instance messages load.
  • load_channel_map_instances: Exact ChannelMap instances to load.
Examples:
>>> options = FlowTimeoutHandlerOptions(
...     method_timeout=timedelta(seconds=30),
...     retry=RetryPolicy(maximum_attempts=3),
...     load_channels=(commands,),
... ).on_failure_proceed_to(TimeoutRecovery)
FlowTimeoutHandlerOptions( method_timeout: datetime.timedelta | None = None, heartbeat_timeout: datetime.timedelta | None = None, retry: RetryPolicy | None = None, durability: StepDurability = <StepDurability.DEFAULT: 'default'>, lock_attributes: tuple[AttributeLock, ...] = (), load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = (), load_attribute_map_instances: tuple[AttributeMapLoad, ...] = (), load_channels: tuple[Channel[typing.Any], ...] = (), load_channel_maps: tuple[ChannelMap[typing.Any], ...] = (), load_channel_map_instances: tuple[ChannelMapLoad, ...] = (), _failure_target: Union[type[Step[NoneType]], Step[NoneType], StepRef, NoneType] = None, _failure_options: StepOptions | None = None)
method_timeout: datetime.timedelta | None = None
heartbeat_timeout: datetime.timedelta | None = None
retry: RetryPolicy | None = None
durability: StepDurability = <StepDurability.DEFAULT: 'default'>
lock_attributes: tuple[AttributeLock, ...] = ()
load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = ()
load_attribute_map_instances: tuple[AttributeMapLoad, ...] = ()
load_channels: tuple[Channel[typing.Any], ...] = ()
load_channel_maps: tuple[ChannelMap[typing.Any], ...] = ()
load_channel_map_instances: tuple[ChannelMapLoad, ...] = ()
def on_failure_proceed_to( self, step: Union[type[Step[NoneType]], Step[NoneType], StepRef], options: StepOptions | None = None) -> FlowTimeoutHandlerOptions:

Return a copy routing exhausted retries to a registered no-input Step.

The recovery Step receives None and reads the final handler failure from Context.recovery_error. Select it by class, registered instance, or StepRef. Use an instance or StepRef when the Flow registers more than one instance of the recovery Step class. The target is validated when Client or SubFlow start options are mapped, not when this method is called.

Arguments:
  • step: Recovery class, registered instance, or StepRef. It must belong to the timed-out Flow and use None input.
  • options: Optional execution overrides for the recovery movement.
Returns:

A new options value with failure routing configured.

Examples:
>>> options = FlowTimeoutHandlerOptions(
...     retry=RetryPolicy(maximum_attempts=1),
... ).on_failure_proceed_to(StepRef("TimeoutRecovery"))
class FlowTimeoutPolicy(enum.Enum):

Control how Dex responds when a positive soft Flow timeout expires.

Attributes:
DEFAULT = <FlowTimeoutPolicy.DEFAULT: 'default'>
FAIL = <FlowTimeoutPolicy.FAIL: 'fail'>
CANCEL = <FlowTimeoutPolicy.CANCEL: 'cancel'>
HANDLER = <FlowTimeoutPolicy.HANDLER: 'handler'>
@dataclass(frozen=True)
class HealthInfo:

Describe one FlowService health-check response.

Attributes:
  • condition: The service-reported health condition.
  • hostname: The hostname of the responding Dex server.
  • duration_seconds: The server-reported response duration in seconds.
HealthInfo(condition: str, hostname: str, duration_seconds: int)
condition: str
hostname: str
duration_seconds: int
class IdReusePolicy(enum.Enum):

Control whether start_flow may reuse an existing Flow ID.

Attributes:
  • DEFAULT: Use the Dex server default, which rejects every previously used Flow ID.
  • ALLOW_IF_PREVIOUS_FAILED: Reuse only after an unsuccessful closed run.
  • ALLOW_IF_NOT_RUNNING: Reuse after any closed run.
  • ALLOW_TERMINATE_IF_RUNNING: Terminate an active run and start the new run.
  • DISALLOW: Reject every previously used Flow ID.
DEFAULT = <IdReusePolicy.DEFAULT: 'default'>
ALLOW_IF_PREVIOUS_FAILED = <IdReusePolicy.ALLOW_IF_PREVIOUS_FAILED: 'allow_if_previous_failed'>
ALLOW_IF_NOT_RUNNING = <IdReusePolicy.ALLOW_IF_NOT_RUNNING: 'allow_if_not_running'>
ALLOW_TERMINATE_IF_RUNNING = <IdReusePolicy.ALLOW_TERMINATE_IF_RUNNING: 'allow_terminate_if_running'>
DISALLOW = <IdReusePolicy.DISALLOW: 'disallow'>
class InvalidStepResultError(dex.FlowDefinitionError):

Identify a Step or RPC handler result that violates the SDK contract.

Attributes:
  • flow_type: The containing Flow type.
  • step_type: The Step type, or None for an RPC.
  • method: The handler method that returned the invalid value.
  • detail: A precise description of the contract violation.
InvalidStepResultError( flow_type: str, step_type: str | None, method: Literal['wait_for', 'execute', 'rpc'], detail: str)

Create an invalid-result definition error.

Arguments:
  • flow_type: The containing Flow type.
  • step_type: The Step type, or None for an RPC.
  • method: "wait_for", "execute", or "rpc".
  • detail: A precise description of the invalid result.
flow_type
step_type
method
detail
class IndexType(enum.Enum):

Select how Dex indexes an Attribute for Flow search.

Attributes:
  • KEYWORD: Index an exact string value.
  • FULL_TEXT: Index a string for tokenized full-text matching.
  • KEYWORD_ARRAY: Index every string in a sequence as a keyword.
  • INT: Index a signed 64-bit integer.
  • DOUBLE: Index a finite floating-point number.
  • BOOL: Index a Boolean value.
  • DATETIME: Index a datetime.datetime value.
KEYWORD = <IndexType.KEYWORD: 'keyword'>
FULL_TEXT = <IndexType.FULL_TEXT: 'full_text'>
KEYWORD_ARRAY = <IndexType.KEYWORD_ARRAY: 'keyword_array'>
INT = <IndexType.INT: 'int'>
DOUBLE = <IndexType.DOUBLE: 'double'>
BOOL = <IndexType.BOOL: 'bool'>
DATETIME = <IndexType.DATETIME: 'datetime'>
@dataclass(frozen=True)
class JsonCodec(typing.Generic[~ValueT]):

Encode an application type as deterministic JSON text.

encoder first converts the typed value to JSON-compatible data; decoder performs the reverse conversion. JSON is emitted with sorted keys, compact separators, and no NaN or infinity values.

Attributes:
  • type_name: The stable type name used in diagnostics.
  • decoder: A callable from parsed JSON-compatible data to ValueT.
  • encoder: A callable from ValueT to JSON-compatible data.
  • expected_type: Optional runtime type hint validated before encoding.
  • wire_kind: Always WireKind.JSON.
Examples:
>>> codec = JsonCodec("Point", lambda data: Point(**data), vars, Point)
>>> codec.decode(codec.encode(Point(x=1, y=2)))
Point(x=1, y=2)
JsonCodec( type_name: str, decoder: Callable[[Any], ~ValueT], encoder: Callable[[~ValueT], Any] = <function JsonCodec.<lambda>>, expected_type: object | None = None)
type_name: str
decoder: Callable[[Any], ~ValueT]
def encoder(value):
expected_type: object | None = None
wire_kind: WireKind = <WireKind.JSON: 'json'>
def encode(self, value: ~ValueT) -> Value:

Encode a typed value as canonical JSON.

Arguments:
  • value: The application value accepted by encoder.
Returns:

A JSON-kind Value containing compact, sorted JSON text.

Raises:
  • TypeError: If expected_type does not match.
  • ValueError: If the encoded data contains unsupported JSON values.
def decode(self, value: Value) -> ~ValueT:

Decode JSON text with this codec's decoder.

Arguments:
  • value: A JSON-kind Value containing valid JSON text.
Returns:

The application value returned by decoder.

Raises:
  • TypeError: If value is not a JSON Value.
  • ValueError: If its JSON text is malformed.
class LongPollTimeoutError(dex.DexServiceError):

Indicate that a retryable long poll ended before its condition was observed.

class RequestTimeoutError(dex.DexServiceError):

Indicate that a caller-defined request budget expired.

@dataclass(frozen=True)
class PersistenceSchema:

Declare the Attributes, Channels, and Streams owned by a Flow type.

Definitions must have unique names and are immutable after Registry creation.

Attributes:
  • attributes: Singleton and map Attribute definitions.
  • channels: Singleton and map Channel definitions.
  • streams: Best-effort Stream definitions.
Examples:
>>> schema = PersistenceSchema.of(status, balances, approvals, events)
PersistenceSchema( attributes: tuple[typing.Union[Attribute[typing.Any], AttributeMap[typing.Any]], ...] = (), channels: tuple[typing.Union[Channel[typing.Any], ChannelMap[typing.Any]], ...] = (), streams: tuple[Stream[typing.Any], ...] = ())
attributes: tuple[typing.Union[Attribute[typing.Any], AttributeMap[typing.Any]], ...] = ()
channels: tuple[typing.Union[Channel[typing.Any], ChannelMap[typing.Any]], ...] = ()
streams: tuple[Stream[typing.Any], ...] = ()
@staticmethod
def of( *definitions: Union[Attribute[Any], AttributeMap[Any], Channel[Any], ChannelMap[Any], Stream[Any]]) -> PersistenceSchema:

Partition persistence definitions into Attributes, Channels, and Streams.

Arguments:
  • *definitions: Attribute, AttributeMap, Channel, ChannelMap, or Stream definitions.
Returns:

A schema with each category kept in argument order.

Raises:
  • TypeError: If a value is not a supported persistence definition.
@dataclass(frozen=True)
class RPCResult(typing.Generic[~OutputT]):

Return typed RPC output, next-Step movements, and Step cancellation.

Attributes:
  • output: The value decoded into the Client caller's expected output type.
  • next_steps: Step movements applied atomically with handler persistence writes.
  • canceling_steps: Flow-wide Step types canceled before next Steps are queued. Each entry is a class, registered instance, or StepRef; narrow it before use. The Worker validates entries when it applies this result, and an invalid entry raises FlowDefinitionError.
RPCResult( output: ~OutputT, next_steps: tuple[StepMovement[typing.Any], ...] = (), canceling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = ())
output: ~OutputT
next_steps: tuple[StepMovement[typing.Any], ...] = ()
canceling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = ()
def with_canceling_steps( self, *steps: Union[type[Step[Any]], Step[Any], StepRef]) -> RPCResult[~OutputT]:

Return a result canceling all current executions of selected Step types.

Dex resolves the selection after RPC persistence commits and before next_steps. Finished, already-canceled, and absent executions are no-ops. RPCs cannot select siblings because an RPC has no Step execution lineage. The Worker validates each target when it applies the returned result.

Arguments:
  • *steps: Registered instances, StepRef values, or unambiguous Step classes.
Returns:

A new RPCResult containing the combined cancellation selectors.

@dataclass(frozen=True)
class RPCInvokeOptions:

Add runtime-selected map instances to one RPC invocation.

Dex unions these selections with the RPC decorator's fixed locks and state loads, then sorts and deduplicates their physical names. Locking and loading are independent. A read-modify-write handler must include the same AttributeMap instance in both collections.

Attributes:
  • lock_attribute_map_instances: AttributeMap instance locks for this call.
  • load_attribute_map_instances: Exact AttributeMap instance snapshots.
  • load_channel_map_instances: Exact ChannelMap pending-message snapshots.
Examples:
>>> options = RPCInvokeOptions(
...     lock_attribute_map_instances=(profiles.lock(partition),),
...     load_attribute_map_instances=(profiles.load(partition),),
... )
>>> client.invoke_rpc(directory.upsert_customer_profile, flow_id, profile,
...                   options=options)
RPCInvokeOptions( lock_attribute_map_instances: tuple[AttributeLock, ...] = (), load_attribute_map_instances: tuple[AttributeMapLoad, ...] = (), load_channel_map_instances: tuple[ChannelMapLoad, ...] = ())
lock_attribute_map_instances: tuple[AttributeLock, ...] = ()
load_attribute_map_instances: tuple[AttributeMapLoad, ...] = ()
load_channel_map_instances: tuple[ChannelMapLoad, ...] = ()
@dataclass(frozen=True)
class RecoveryErrorInfo:

Describe the final failure that selected a recovery path.

Attributes:
  • detail: Failure message supplied by the Worker or backend.
  • error_type: Original Worker error type or backend failure type.
  • attempt_progress: Furthest progress across all attempts of this Step method. Defaults to StepAttemptProgress.UNSPECIFIED for older records and for construction without progress. Treat that default like NOT_COMPLETED. Only NOT_STARTED guarantees the handler did not run. See StepAttemptProgress for a recovery example.
RecoveryErrorInfo( detail: str, error_type: str, attempt_progress: StepAttemptProgress = <StepAttemptProgress.UNSPECIFIED: 'unspecified'>)
detail: str
error_type: str
attempt_progress: StepAttemptProgress = <StepAttemptProgress.UNSPECIFIED: 'unspecified'>
class StepAttemptProgress(enum.Enum):

Describe the furthest progress across attempts of a failed Step method.

Inspect Context.recovery_error.attempt_progress before deciding whether recovery may safely repeat a side effect. Only NOT_STARTED guarantees the handler did not run. Other values require idempotent recovery or an application check for completed side effects.

Attributes:
  • UNSPECIFIED: An older Server or a failure recorded before upgrade supplied no progress. Treat it like NOT_COMPLETED because an attempt may have run.
  • NOT_STARTED: The handler did not run in any attempt.
  • NOT_COMPLETED: Some attempt may have run, but none completed.
Examples:
>>> error = RecoveryErrorInfo("unavailable", "dispatch", StepAttemptProgress.NOT_STARTED)
>>> can_repeat_side_effect = error.attempt_progress is StepAttemptProgress.NOT_STARTED
UNSPECIFIED = <StepAttemptProgress.UNSPECIFIED: 'unspecified'>
NOT_STARTED = <StepAttemptProgress.NOT_STARTED: 'not_started'>
NOT_COMPLETED = <StepAttemptProgress.NOT_COMPLETED: 'not_completed'>
@dataclass(frozen=True)
class RetryAfterError(builtins.Exception):

Requests a delay before Dex retries a failed worker method.

Raise an instance with retry_after() from a Step method when the failure is retryable but the next attempt should happen at a known time. Dex reports the original cause as the worker failure and uses the requested delay instead of the retry policy's calculated interval.

Attributes:
  • after_seconds (int): Positive delay in seconds before the next retry.
  • cause (BaseException): Original failure reported to Dex and re-raised as the retry cause.
RetryAfterError(after_seconds: int, cause: BaseException)
after_seconds: int
cause: BaseException
class RpcLockConflictError(dex.DexServiceError):

Indicate that an RPC could not acquire all requested Attribute locks.

@dataclass(frozen=True)
class Registry:

Validate and store Flow definitions shared by Client and Worker.

Registry construction inspects Step and RPC signatures, resolves codecs, checks names and locks, and assembles Attribute indexes atomically. The result is safe for concurrent reads.

Attributes:
  • flows: The registered Flow instances in input order.
  • codec_registry: The codec registry used for every handler and persistence type.
Examples:
>>> orders = Orders()
>>> registry = Registry([orders], CodecRegistry({Money: money_codec}))
Registry( flows: Sequence[Flow[Any]], codec_registry: CodecRegistry | None = None, *, allow_async_handlers: bool = False)

Validate and assemble a set of Flow definitions.

Arguments:
  • flows: Flow instances with unique Flow types.
  • codec_registry: Optional custom codecs; defaults to CodecRegistry().
  • allow_async_handlers: Accept async def Step and RPC handlers. Async Steps annotate AsyncContext; async RPCs and timeout handlers keep Context. Sync Client/Worker construction leaves this False.
Raises:
  • FlowDefinitionError: If any Flow, Step, RPC, persistence definition, annotation, codec, lock, or index is invalid.
flows: tuple[Flow[typing.Any], ...]
codec_registry: CodecRegistry
@staticmethod
def physical_name(name: str, instance: str) -> str:

Return the physical name for a map instance.

Arguments:
  • name: The AttributeMap or ChannelMap name.
  • instance: The map instance. Slash is prohibited because it is a reserved character.
Returns:

The physical name.

Raises:
  • ValueError: If instance is empty or contains /.
@dataclass(frozen=True)
class TimeTravelOptions:

Configure creation of a new run from existing Flow history.

Exactly one selector matching type must be set. The Client validates combinations before sending the request.

Attributes:
  • type: The time travel point selector kind.
  • history_event_time: Timestamp used by HISTORY_EVENT_TIME.
  • step_type: Registered Step type used by STEP_TYPE.
  • step_execution_id: Exact execution ID used by STEP_EXECUTION_ID.
  • step_method: WaitFor or Execute boundary required by STEP_EXECUTION_ID.
  • reason: Optional operator-readable time travel reason.
  • skip_writes_reapply: Do not replay RPCs, Channel publications, or Attribute writes after the selected point.
TimeTravelOptions( type: TimeTravelType, history_event_time: datetime.datetime | None = None, step_type: str | None = None, step_execution_id: str | None = None, step_method: TimeTravelStepMethod | None = None, reason: str | None = None, skip_writes_reapply: bool = False)
history_event_time: datetime.datetime | None = None
step_type: str | None = None
step_execution_id: str | None = None
step_method: TimeTravelStepMethod | None = None
reason: str | None = None
skip_writes_reapply: bool = False
class TimeTravelStepMethod(enum.Enum):

Select the Step method used as a Step execution time travel boundary.

Attributes:
  • WAIT_FOR: Rerun WaitFor and everything after it.
  • EXECUTE: Keep the WaitFor result and rerun Execute and everything after it.
WAIT_FOR = <TimeTravelStepMethod.WAIT_FOR: 'wait_for'>
EXECUTE = <TimeTravelStepMethod.EXECUTE: 'execute'>
class TimeTravelType(enum.Enum):

Select the historical point from which time travel should create a new run.

Attributes:
  • BEGINNING: Restart from the beginning of the Flow.
  • HISTORY_EVENT_TIME: Resume at the first event at or after a timestamp.
  • STEP_TYPE: Resume at the latest execution of a Step type.
  • STEP_EXECUTION_ID: Resume at one exact Step execution ID.
BEGINNING = <TimeTravelType.BEGINNING: 'beginning'>
HISTORY_EVENT_TIME = <TimeTravelType.HISTORY_EVENT_TIME: 'history_event_time'>
STEP_TYPE = <TimeTravelType.STEP_TYPE: 'step_type'>
STEP_EXECUTION_ID = <TimeTravelType.STEP_EXECUTION_ID: 'step_execution_id'>
@dataclass(frozen=True)
class RetryPolicy:

Configure exponential retries for a Step handler or Flow.

A None duration or zero numeric value uses the server default. The default total duration is four hours. Attempts include the initial call. Retries stop at the first configured limit. Step total duration starts when the Step is first handed to a Worker. After an attempt fails, Dex uses the elapsed duration to decide whether to schedule another retry. It never ends a running attempt.

Attributes:
  • initial_interval: Delay before the first retry.
  • backoff_coefficient: Interval multiplier; zero uses the server default.
  • maximum_interval: Upper bound for one retry delay.
  • maximum_attempts: Total attempt limit; zero uses the server default.
  • total_duration: Retry scheduling budget counted from the first hand-out to a Worker. Checked only after an attempt fails; it does not end a running attempt. None uses four hours.
RetryPolicy( initial_interval: datetime.timedelta | None = None, backoff_coefficient: float = 0.0, maximum_interval: datetime.timedelta | None = None, maximum_attempts: int = 0, total_duration: datetime.timedelta | None = None)
initial_interval: datetime.timedelta | None = None
backoff_coefficient: float = 0.0
maximum_interval: datetime.timedelta | None = None
maximum_attempts: int = 0
total_duration: datetime.timedelta | None = None
@dataclass(frozen=True)
class SearchFlowEntry:

Represent one indexed Flow run returned by search_flows.

Attributes:
  • flow_id: The stable application-assigned Flow ID.
  • run_id: The server-assigned run ID for this result.
  • flow_type: The registered Flow type name.
  • status: The indexed run status.
  • started_at: The timezone-aware start timestamp.
  • closed_at: The close timestamp, or None for open or unknown runs.
  • indexed_attributes: Hydrated Values keyed by physical search-index name.
SearchFlowEntry( flow_id: str, run_id: str, flow_type: str, status: FlowStatus, started_at: datetime.datetime, closed_at: datetime.datetime | None, indexed_attributes: Mapping[str, Value])
flow_id: str
run_id: str
flow_type: str
status: FlowStatus
started_at: datetime.datetime
closed_at: datetime.datetime | None
indexed_attributes: Mapping[str, Value]
@dataclass(frozen=True)
class SearchFlowsPage:

Contain one server-ordered page of Flow search results.

Attributes:
  • flows: The result entries in server-defined query order.
  • next_page_token: An opaque token for the next request, or "" on the final page.
SearchFlowsPage(flows: Sequence[SearchFlowEntry], next_page_token: str)
flows: Sequence[SearchFlowEntry]
next_page_token: str
class StepCompletion:

Contain one output-bearing Step completion returned by wait_for_flow.

Attributes:
  • step_type: The registered Step type that produced the output.
  • step_execution_id: The exact server Step execution identity.
StepCompletion( completion: dex_pb2.StepCompletionOutput, decoder: Callable[[dex_pb2.Value, type[typing.Any]], typing.Any])

Create a completion around an already hydrated wire value.

Arguments:
  • completion: The hydrated completion message returned by Dex.
  • decoder: The SDK value decoder used by decode().
step_type: str

Return the registered Step type that produced this output.

Returns:

The registered Step type.

step_execution_id: str

Return the exact server Step execution identity.

Returns:

The server-assigned Step execution ID.

def decode(self, output_type: type[~OutputT]) -> ~OutputT:

Decode this hydrated Step output as output_type.

Arguments:
  • output_type: The concrete Python type expected by the caller.
Returns:

The decoded Step output.

Raises:
  • ValueMappingError: If the output cannot be decoded as output_type.
@dataclass(frozen=True)
class StartFlowOptions:

Configure creation of a new Flow execution.

Durations are datetime.timedelta values. None uses the registered Flow or server default. Use with_attribute() to add initial Attribute writes.

To retry one logical start safely, for example after a lost response, set a stable request_id and ignore_already_started=True together. Reuse the same values on every retry.

Attributes:
  • timeout: Optional durable soft timeout. None or zero disables it.
  • timeout_policy: Action taken when a positive timeout expires.
  • timeout_handler_options: Execute policy used when timeout policy is HANDLER.
  • start_delay: Optional delay before the starting Step becomes eligible.
  • id_reuse_policy: Flow ID reuse policy; defaults to DEFAULT.
  • retry_policy: Optional Flow-level retry policy.
  • config_override: Optional FlowConfig applied over registered defaults.
  • ignore_already_started: Return the existing run rather than raising FlowAlreadyStartedError when supported by the server. It applies only when request_id matches the request ID that started the existing run. A different or generated request ID still raises FlowAlreadyStartedError.
  • request_id: Optional non-empty idempotency key. None generates one. A retried start returns the existing run only when ignore_already_started is True and this request ID matches the request that started that run. Without that option, the retry raises FlowAlreadyStartedError even when the first attempt created the Flow. The match applies only when the ID reuse policy rejects the start. A policy that permits another run starts a new run instead. Dex compares only request IDs, not input or other options, so never reuse a request ID for a different logical start.
Examples:
>>> options = (
...     StartFlowOptions(timeout=timedelta(hours=1))
...     .with_attribute(status, "pending")
...     .with_attribute(balances, "merchant-7", 1200)
... )
>>> retry_safe_options = StartFlowOptions(
...     ignore_already_started=True,
...     request_id=f"start-order-{order_id}",
... )
>>> run_id = client.start_flow(
...     order_flow, f"order-{order_id}", order_input, retry_safe_options
... )
StartFlowOptions( timeout: datetime.timedelta | None = None, timeout_policy: FlowTimeoutPolicy = <FlowTimeoutPolicy.DEFAULT: 'default'>, timeout_handler_options: FlowTimeoutHandlerOptions | None = None, start_delay: datetime.timedelta | None = None, id_reuse_policy: IdReusePolicy = <IdReusePolicy.DEFAULT: 'default'>, retry_policy: RetryPolicy | None = None, _attribute_initializations: tuple[dex.flow_options._AttributeInitialization, ...] = (), config_override: FlowConfig | None = None, ignore_already_started: bool = False, request_id: str | None = None)
timeout: datetime.timedelta | None = None
timeout_policy: FlowTimeoutPolicy = <FlowTimeoutPolicy.DEFAULT: 'default'>
timeout_handler_options: FlowTimeoutHandlerOptions | None = None
start_delay: datetime.timedelta | None = None
id_reuse_policy: IdReusePolicy = <IdReusePolicy.DEFAULT: 'default'>
retry_policy: RetryPolicy | None = None
config_override: FlowConfig | None = None
ignore_already_started: bool = False
request_id: str | None = None
def with_attribute( self, attribute: Union[Attribute[Any], AttributeMap[Any]], /, *args: object) -> StartFlowOptions:

Return a copy with one initial Attribute value appended.

Singleton Attributes take (attribute, value); AttributeMaps take (attribute, instance, value). Calls preserve initialization order.

Arguments:
  • attribute: A registered Attribute or AttributeMap definition.
  • *args: The value, or the AttributeMap instance followed by its value.
Returns:

A new options object containing the additional initialization.

Raises:
  • TypeError: If arguments do not match either supported form.
  • ValueError: If an AttributeMap instance is empty or contains /.
class SubFlow:

Create durable SubFlow Conditions and read their Execute results.

@staticmethod
def run( flow: Flow[typing.Any], input: object, options: SubFlowOptions | None = None) -> dex.condition.Condition:

Create a Condition that starts or reuses flow and awaits completion.

Arguments:
  • flow: The exact target Flow instance registered by the Worker.
  • input: Input accepted by the target Flow's starting Step.
  • options: Optional reuse, timing, Attribute, configuration, and ID settings.
Returns:

A durable Condition accepted by dex.Wait.

@staticmethod
def get_condition_results( context: Context, index: int = 0) -> FlowResult:

Return one stable-indexed SubFlow result during Step execute.

Arguments:
  • context: The current Dex Execute context.
  • index: Zero-based SubFlow order within the surrounding Wait.
Returns:

A terminal result or an AnyOf loser's running snapshot.

Raises:
  • TypeError: If context is not managed by Dex.
  • ValueError: If called outside Execute or index is unavailable.
@staticmethod
def get_flow_id(context: Context, index: int = 0) -> str:

Return the generated Flow ID for one SubFlow Condition.

The ID remains usable after an AnyOf winner completes and may be passed to Client.stop_flow to stop a still-running loser.

Arguments:
  • context: The current Dex Execute context.
  • index: Zero-based SubFlow order within the surrounding Wait.
Returns:

The stable server-generated SubFlow Flow ID.

@dataclass(frozen=True)
class SubFlowOptions:

Configure one durable SubFlow Condition.

The SubFlow inherits the parent's effective Flow configuration. None values preserve normal start defaults. Dex generates the Flow ID and request ID.

Attributes:
  • timeout: Optional maximum SubFlow lifetime.
  • timeout_policy: Action taken when a positive timeout expires.
  • timeout_handler_options: Execute policy used when timeout policy is HANDLER.
  • start_delay: Optional delay before its starting Step.
  • retry_policy: Optional whole-Flow retry policy.
  • config_override: Fields applied over the inherited parent configuration.
  • reuse_policy: Existing-execution resolution policy.
  • condition_id: Stable ID required by Wait.any_combination_of.
SubFlowOptions( timeout: datetime.timedelta | None = None, timeout_policy: FlowTimeoutPolicy = <FlowTimeoutPolicy.DEFAULT: 'default'>, timeout_handler_options: FlowTimeoutHandlerOptions | None = None, start_delay: datetime.timedelta | None = None, retry_policy: RetryPolicy | None = None, _attribute_initializations: tuple[dex.flow_options._AttributeInitialization, ...] = (), config_override: FlowConfig | None = None, reuse_policy: SubFlowReusePolicy = <SubFlowReusePolicy.RESTART_IF_PREVIOUS_EXITS_ABNORMALLY: 'restart_if_previous_exits_abnormally'>, condition_id: str | None = None)
timeout: datetime.timedelta | None = None
timeout_policy: FlowTimeoutPolicy = <FlowTimeoutPolicy.DEFAULT: 'default'>
timeout_handler_options: FlowTimeoutHandlerOptions | None = None
start_delay: datetime.timedelta | None = None
retry_policy: RetryPolicy | None = None
config_override: FlowConfig | None = None
reuse_policy: SubFlowReusePolicy = <SubFlowReusePolicy.RESTART_IF_PREVIOUS_EXITS_ABNORMALLY: 'restart_if_previous_exits_abnormally'>
condition_id: str | None = None
def with_attribute( self, attribute: Union[Attribute[Any], AttributeMap[Any]], /, *args: object) -> SubFlowOptions:

Return a copy with one target-Flow Attribute initialization appended.

Arguments:
  • attribute: A singleton Attribute or AttributeMap owned by the SubFlow.
  • *args: Its value, or a map instance followed by its value.
Returns:

A new immutable options value.

Raises:
  • TypeError: If arguments do not match either supported form.
  • ValueError: If an AttributeMap instance is empty or contains /.
class SubFlowReusePolicy(enum.Enum):

Control how a generated SubFlow Flow ID resolves an existing execution.

Attributes:
  • ATTACH: Attach to a running execution or return its terminal result.
  • RESTART_IF_PREVIOUS_EXITS_ABNORMALLY: Attach while running, return a successful result, and restart an abnormal previous execution.
  • ALWAYS_RESTART: Replace any different existing execution.
ATTACH = <SubFlowReusePolicy.ATTACH: 'attach'>
RESTART_IF_PREVIOUS_EXITS_ABNORMALLY = <SubFlowReusePolicy.RESTART_IF_PREVIOUS_EXITS_ABNORMALLY: 'restart_if_previous_exits_abnormally'>
ALWAYS_RESTART = <SubFlowReusePolicy.ALWAYS_RESTART: 'always_restart'>
@dataclass(frozen=True)
class StepExecutionId:

Identify one numbered execution of a Step type.

Attributes:
  • step_type: The registered Step type name.
  • number: The positive execution number; defaults to the first execution.
StepExecutionId(step_type: str, number: int = 1)
step_type: str
number: int = 1
@dataclass(frozen=True)
class StepDecision:

Describe the durable state transition returned by Step.execute.

Prefer the module-level factory functions instead of constructing this class directly; they populate the fields required for each decision kind.

Attributes:
  • kind: The transition behavior interpreted by Dex.
  • movements: Destination Steps for a next decision.
  • output: Flow completion value, or an internal marker when omitted.
  • reason: Human-readable force-failure reason.
  • empty_channels: Channels that must be empty for conditional completion.
  • fallback: Movement used when conditional completion cannot proceed.
  • canceling_steps: Flow-wide Step type cancellation selectors. Each entry is a class, registered instance, or StepRef; narrow it before use.
  • canceling_sibling_steps: Same-source Step type cancellation selectors with the same entry types as canceling_steps.
StepDecision( kind: dex.step.DecisionKind, movements: tuple[StepMovement[typing.Any], ...] = (), output: object = <object object>, reason: str = '', empty_channels: tuple[typing.Union[Channel[typing.Any], ChannelMap[typing.Any]], ...] = (), fallback: Optional[StepMovement[Any]] = None, canceling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = (), canceling_sibling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = ())
kind: dex.step.DecisionKind
movements: tuple[StepMovement[typing.Any], ...] = ()
output: object = <object object>
reason: str = ''
empty_channels: tuple[typing.Union[Channel[typing.Any], ChannelMap[typing.Any]], ...] = ()
fallback: Optional[StepMovement[Any]] = None
canceling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = ()
canceling_sibling_steps: tuple[typing.Union[type[Step[typing.Any]], Step[typing.Any], StepRef], ...] = ()
def with_canceling_steps( self, *steps: Union[type[Step[Any]], Step[Any], StepRef]) -> StepDecision:

Return a decision canceling all current executions of selected Step types.

Dex resolves one snapshot after this Execute succeeds. Finished, already-canceled, and absent executions are no-ops. Steps scheduled by this decision are excluded. Repeated calls take the union, and Flow-wide selection supersedes sibling selection. The Worker validates each target when it applies the returned decision. An invalid target raises FlowDefinitionError at that point.

Arguments:
  • *steps: Registered instances, StepRef values, or unambiguous Step classes.
Returns:

A new decision containing the combined Flow-wide selectors.

def with_canceling_sibling_steps( self, *steps: Union[type[Step[Any]], Step[Any], StepRef]) -> StepDecision:

Return a decision canceling selected same-source Step executions.

A sibling has the same Context.from_step_execution_id as the execution returning this decision. Snapshot, no-op, and validation behavior match with_canceling_steps. Flow-wide selection wins for the same Step type.

Arguments:
  • *steps: Registered instances, StepRef values, or unambiguous Step classes.
Returns:

A new decision containing the combined sibling selectors.

class Step(typing.Generic[~InputT], abc.ABC):

Define one typed unit of durable Flow application logic.

Subclasses implement execute and may implement wait_for. Handler methods must remain deterministic with respect to their inputs and Context state; external side effects belong behind idempotent application integrations.

Examples:
>>> class Charge(Step[ChargeInput]):
...     def execute(self, context: Context, input: ChargeInput) -> StepDecision:
...         status.set(context, "charged")
...         return graceful_complete(input.order_id)
@abstractmethod
def execute( self, context: Context, input: ~InputT) -> Union[StepDecision, Generator[StepOutput, NoneType, StepDecision]]:

Execute application logic and produce the next durable decision.

An ordinary handler returns StepDecision directly. A progress handler yields heartbeat and Stream StepOutput values, then returns StepDecision as its generator return value.

Arguments:
  • context: Execution metadata and decision-local persistence operations.
  • input: The decoded value passed by the incoming Step movement.
Returns:

A StepDecision, or a generator that yields StepOutput and returns the StepDecision describing movements or Flow completion.

Raises:
  • Exception: Application exceptions are retried according to StepOptions and may ultimately fail or reroute the Flow.
def wait_for( self, context: Context, input: ~InputT) -> Union[Wait, Generator[StepOutput, NoneType, Wait]]:

Describe durable conditions that must be ready before execute.

Override this method only when the Step waits. Dex skips the base implementation and invokes execute immediately. A progress handler yields heartbeat and Stream StepOutput values, then returns Wait as its generator return value.

Arguments:
  • context: Execution metadata and decision-local persistence operations.
  • input: The same decoded input later passed to execute.
Returns:

A durable Wait condition tree, or a generator that yields StepOutput and returns that Wait.

def get_step_type(self) -> str:

Return the name used to register and address this Step.

Override it to decouple the protocol Step type from the Python class name.

Returns:

A non-empty name unique within the containing Flow.

def get_input_type(self) -> object | None:

Return the input type that overrides this Step's handler annotations.

Override it in a generic Step whose input annotation is a TypeVar, so each registered instance can declare its concrete input type. Registry calls it during registration and resolves the returned type through its CodecRegistry for both execute and wait_for. Return type(None) for a Step that receives no input value. Context and return annotations remain required.

Returns:

The input type for this instance, or None to infer it from the handler input annotations. The default returns None.

Raises:
  • FlowDefinitionError: Raised by Registry, not by this method, when the CodecRegistry has no codec for the returned type.
Examples:
>>> class Relay(Step[InputT], Generic[InputT]):
...     def __init__(self, step_type: str, input_type: type[InputT]) -> None:
...         self._step_type = step_type
...         self._input_type = input_type
...
...     def get_step_type(self) -> str:
...         return self._step_type
...
...     def get_input_type(self) -> object | None:
...         return self._input_type
...
...     def execute(self, context: Context, input: InputT) -> StepDecision:
...         return graceful_complete(input)
>>> numbers = Relay("RelayNumber", int)
>>> texts = Relay("RelayText", str)
def get_step_options(self) -> StepOptions | None:

Return this Step's static execution options.

Returns:

Step-specific options, or None to use Flow and server defaults.

@dataclass(frozen=True)
class StepList(typing.Generic[~StartT]):

Build the ordered Step definitions for a Flow.

A Flow may have zero or one starting Step. Other Steps are reachable only through decisions, RPC results, or failure routing.

StepList(_definitions: tuple[dex.step._StepDef, ...])
@classmethod
def empty(cls) -> StepList[~StartT]:

Create a StepList with no starting or other Steps.

Returns:

An empty StepList.

@staticmethod
def start_step(step: Step[~StartT]) -> StepList[~StartT]:

Create a StepList whose first definition is the Flow starting Step.

Arguments:
  • step: The Step receiving start_flow input.
Returns:

A StepList with exactly one starting Step.

@classmethod
def without_start_step(cls, *steps: Step[typing.Any]) -> StepList[~StartT]:

Create a StepList with only non-starting Steps.

Arguments:
  • *steps: Registered Steps reachable through other entry paths.
Returns:

A StepList that requires None start input.

def other_steps(self, *steps: Step[typing.Any]) -> StepList[~StartT]:

Return a copy with additional non-starting Steps appended.

Arguments:
  • *steps: Steps to append in registration order.
Returns:

A new StepList.

class StepDurability(enum.Enum):

Control when a Step handler result is durably acknowledged.

ASYNC gives lower latency and faster single Flows. Sustained throughput is about the same as SYNC.

Attributes:
  • DEFAULT: Use FlowConfig, then the server's synchronous default.
  • SYNC: Persist the result before acknowledging the Worker call.
  • ASYNC: Allow acknowledgement before persistence completes.
DEFAULT = <StepDurability.DEFAULT: 'default'>
SYNC = <StepDurability.SYNC: 'sync'>
ASYNC = <StepDurability.ASYNC: 'async'>
@dataclass(frozen=True)
class StepMovement(typing.Generic[~InputT]):

Schedule one registered Step with typed input and optional options.

Code that reads step must narrow it before using class members, because it may hold a class, a registered instance, or a StepRef.

Attributes:
  • step: Destination class, registered instance, or StepRef in this Flow.
  • input: The value decoded by the destination Step's input codec.
  • options: Optional per-movement StepOptions override.
StepMovement( step: Union[type[Step[~InputT]], Step[~InputT], StepRef], input: ~InputT, options: StepOptions | None = None)
step: Union[type[Step[~InputT]], Step[~InputT], StepRef]
input: ~InputT
options: StepOptions | None = None
@staticmethod
def of( step: Union[type[Step[~InputT]], Step[~InputT], StepRef], input: ~InputT, *, options: StepOptions | None = None) -> StepMovement[~InputT]:

Create a typed movement to a Step.

The Worker validates the target when it applies the handler result that returns this movement. An unknown StepRef, ambiguous class, or unregistered instance raises FlowDefinitionError at that point.

Arguments:
  • step: Destination class, registered instance, or StepRef. A class must be unambiguous.
  • input: A value compatible with the Step's input annotation.
  • options: Optional per-movement options.
Returns:

A StepMovement.

class StepOutput:

Represent one progress frame yielded by a synchronous Step handler.

Create outputs with heartbeat() or dex.stream.Stream.write(). A synchronous generator yields these frames and returns its final Wait or StepDecision through the generator return value.

StepOutput(*args: object, **kwargs: object)

Reject direct construction of the abstract output marker.

@dataclass(frozen=True)
class StepRef:

Address a registered Step by its stable Step type without an implementation.

Use this reference with go_to, StepMovement, failure routing, or cancellation. It cannot execute or appear in a StepList. The constructor never checks that the Step type exists. Registry validates execute-failure routes declared by get_step_options. Client validates timeout-handler failure routes when it maps start options. A StepRef returned from a handler, in a movement or cancellation selection, is validated only when the Worker applies that returned result. Unknown types raise FlowDefinitionError. Input is checked by the destination's codec rather than inferred from the reference.

Attributes:
  • step_type: Non-empty Step type returned by the destination's get_step_type.
Examples:
>>> movement = StepMovement.of(StepRef("Charge"), 42)
StepRef(step_type: str)
step_type: str
StepTarget = typing.Union[type[Step[~InputT]], Step[~InputT], StepRef]
@dataclass(frozen=True)
class StepOptions:

Configure timeouts, retries, durability, locks, and failure routing for a Step.

Fields set to None or DEFAULT use Flow or server policy. Attribute locks are acquired separately for wait_for and execute and must reference definitions in the containing Flow's PersistenceSchema. Asynchronous durability gives lower latency and faster single Flows. Sustained throughput is about the same as synchronous durability.

Attributes:
  • wait_for_method_timeout: Maximum duration of one wait_for attempt; None uses two hours.
  • execute_method_timeout: Maximum duration of one execute attempt; None uses two hours.
  • heartbeat_timeout: Handler progress deadline; None or zero uses one minute. The server-configured explicit minimum defaults to ten seconds.
  • wait_for_retry: Optional retry policy for wait_for.
  • execute_retry: Optional retry policy for execute.
  • wait_for_failure: Behavior after all wait_for attempts fail.
  • wait_for_durability: Durability used for the wait_for result.
  • execute_durability: Durability used for the execute result.
  • wait_for_lock_attributes: Attribute locks held during wait_for.
  • execute_lock_attributes: Attribute locks held during execute.
  • wait_for_load_attribute_maps: AttributeMaps fully loaded for wait_for.
  • wait_for_load_attribute_map_instances: Exact AttributeMap instances loaded for wait_for.
  • wait_for_load_channels: Channels whose pending messages load for wait_for.
  • wait_for_load_channel_maps: ChannelMaps fully loaded for wait_for.
  • wait_for_load_channel_map_instances: Exact ChannelMap instances loaded for wait_for.
  • execute_load_attribute_maps: AttributeMaps fully loaded for execute.
  • execute_load_attribute_map_instances: Exact AttributeMap instances loaded for execute.
  • execute_load_channels: Channels whose pending messages load for execute.
  • execute_load_channel_maps: ChannelMaps fully loaded for execute.
  • execute_load_channel_map_instances: Exact ChannelMap instances loaded for execute.
StepOptions( wait_for_method_timeout: datetime.timedelta | None = None, execute_method_timeout: datetime.timedelta | None = None, heartbeat_timeout: datetime.timedelta | None = None, wait_for_retry: RetryPolicy | None = None, execute_retry: RetryPolicy | None = None, wait_for_failure: WaitForFailurePolicy = <WaitForFailurePolicy.FAIL_FLOW: 'fail_flow'>, wait_for_durability: StepDurability = <StepDurability.DEFAULT: 'default'>, execute_durability: StepDurability = <StepDurability.DEFAULT: 'default'>, wait_for_lock_attributes: tuple[AttributeLock, ...] = (), execute_lock_attributes: tuple[AttributeLock, ...] = (), wait_for_load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = (), wait_for_load_attribute_map_instances: tuple[AttributeMapLoad, ...] = (), wait_for_load_channels: tuple[Channel[typing.Any], ...] = (), wait_for_load_channel_maps: tuple[ChannelMap[typing.Any], ...] = (), wait_for_load_channel_map_instances: tuple[ChannelMapLoad, ...] = (), execute_load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = (), execute_load_attribute_map_instances: tuple[AttributeMapLoad, ...] = (), execute_load_channels: tuple[Channel[typing.Any], ...] = (), execute_load_channel_maps: tuple[ChannelMap[typing.Any], ...] = (), execute_load_channel_map_instances: tuple[ChannelMapLoad, ...] = (), _execute_failure_target: Union[type[Step[Any]], Step[Any], StepRef, NoneType] = None, _execute_failure_options: StepOptions | None = None)
wait_for_method_timeout: datetime.timedelta | None = None
execute_method_timeout: datetime.timedelta | None = None
heartbeat_timeout: datetime.timedelta | None = None
wait_for_retry: RetryPolicy | None = None
execute_retry: RetryPolicy | None = None
wait_for_failure: WaitForFailurePolicy = <WaitForFailurePolicy.FAIL_FLOW: 'fail_flow'>
wait_for_durability: StepDurability = <StepDurability.DEFAULT: 'default'>
execute_durability: StepDurability = <StepDurability.DEFAULT: 'default'>
wait_for_lock_attributes: tuple[AttributeLock, ...] = ()
execute_lock_attributes: tuple[AttributeLock, ...] = ()
wait_for_load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = ()
wait_for_load_attribute_map_instances: tuple[AttributeMapLoad, ...] = ()
wait_for_load_channels: tuple[Channel[typing.Any], ...] = ()
wait_for_load_channel_maps: tuple[ChannelMap[typing.Any], ...] = ()
wait_for_load_channel_map_instances: tuple[ChannelMapLoad, ...] = ()
execute_load_attribute_maps: tuple[AttributeMap[typing.Any], ...] = ()
execute_load_attribute_map_instances: tuple[AttributeMapLoad, ...] = ()
execute_load_channels: tuple[Channel[typing.Any], ...] = ()
execute_load_channel_maps: tuple[ChannelMap[typing.Any], ...] = ()
execute_load_channel_map_instances: tuple[ChannelMapLoad, ...] = ()
def on_execute_failure_proceed_to( self, step: Union[type[Step[Any]], Step[Any], StepRef], options: StepOptions | None = None) -> StepOptions:

Return a copy that routes exhausted execute retries to another Step.

The fallback Step receives the failing Step's input, so Registry rejects a fallback whose input type differs. The check is skipped for Any, object, and unresolved TypeVar input types.

Arguments:
  • step: Fallback class, registered instance, or StepRef; validated at registration.
  • options: Optional options applied to the fallback movement.
Returns:

A new StepOptions object with failure routing configured.

@dataclass(frozen=True)
class Stream(typing.Generic[~ValueT]):

Define a typed best-effort resumable Stream owned by one Flow type.

Attributes:
  • name: Logical name unique within the Flow persistence schema.
  • value_type: Python type encoded for every message.
  • stream_capacity_bytes: Positive approximate budget shared by all Flow instances.
Stream(name: str, value_type: type[~ValueT], stream_capacity_bytes: int)
name: str
value_type: type[~ValueT]
stream_capacity_bytes: int
def write( self, context: Context, value: ~ValueT) -> StepOutput | None:

Emit one best-effort message from the current Step execution.

A synchronous generator must yield the returned StepOutput. An asynchronous Step calls this method without await; the message enters the invocation's ordered output queue immediately. Neither form waits for Dex Stream Store persistence. Calls may write the same Stream any number of times. RPC and Flow-timeout Contexts reject Stream writes.

Arguments:
  • context: Current synchronous or asynchronous Step Context.
  • value: Typed message to append.
Returns:

A StepOutput for a synchronous Context, or None for AsyncContext.

Raises:
  • ValueError: If the Stream is unregistered or the Context is not a Step.
  • ValueMappingError: If value cannot be encoded.
def buffered_text( self: Stream[str], context: Context, *, flush_interval: datetime.timedelta = datetime.timedelta(seconds=1), max_buffered_bytes: int = 16384) -> AsyncBufferedTextStream | BufferedTextStream:

Create an invocation-managed writer that batches text chunks.

The first non-empty chunk starts a one-shot timer. Async Step writers append their current batch after flush_interval even without another chunk. Synchronous generator writers check the interval only when write is called and require an explicit final flush. Both forms flush early after crossing the soft UTF-8 byte threshold. Chunks are never split or modified.

AsyncWorker finalizes the writer before sending the invocation result or error. Its synchronous write method can be passed directly as a text-delta callback. Neither form waits for Dex Stream Store acknowledgement.

Arguments:
  • context: Current asynchronous or synchronous Step Context.
  • flush_interval: Positive maximum async buffering interval. Defaults to one second.
  • max_buffered_bytes: Positive soft UTF-8 threshold. Defaults to 16 KiB.
Returns:

An async invocation-managed writer, or a cooperative synchronous writer.

Raises:
  • TypeError: If this Stream does not carry str values or an option has the wrong type.
  • ValueError: If an option is not positive, the Stream is unregistered, or the Context is not a Step invocation.
@dataclass(frozen=True)
class StreamMessage(typing.Generic[~ValueT]):

Describe one retained Stream message returned by a Client.

Attributes:
  • value: Decoded application message.
  • resume_token: Opaque token for the next read.
  • created_time: Server-assigned UTC creation time.
  • source: Informational client source or Step-generated #stepExecutionID.
StreamMessage( value: ~ValueT, resume_token: str, created_time: datetime.datetime, source: str)
value: ~ValueT
resume_token: str
created_time: datetime.datetime
source: str
@dataclass(frozen=True)
class StreamMessagesPage(typing.Generic[~ValueT]):

Contain one newest-first page of retained Stream messages.

Pass next_page_token unchanged to the next Client list call. An empty token marks the final page. Stream trimming may remove messages between pages.

Attributes:
  • messages: Retained messages in newest-first order.
  • next_page_token: Opaque token for the next older page, or "".
StreamMessagesPage( messages: Sequence[StreamMessage[~ValueT]], next_page_token: str)
messages: Sequence[StreamMessage[~ValueT]]
next_page_token: str
@dataclass(frozen=True)
class StopFlowOptions:

Configure a stop_flow request.

Attributes:
  • type: The stop behavior; defaults to cooperative cancellation.
  • reason: Optional operator-readable reason recorded by Dex.
StopFlowOptions( type: StopType = <StopType.CANCEL: 'cancel'>, reason: str | None = None)
type: StopType = <StopType.CANCEL: 'cancel'>
reason: str | None = None
class StopType(enum.Enum):

Select how an active Flow should close.

Attributes:
  • CANCEL: Request cooperative cancellation.
  • TERMINATE: Force immediate termination.
  • FAIL: Close the Flow as failed with the reason in StopFlowOptions.
CANCEL = <StopType.CANCEL: 'cancel'>
TERMINATE = <StopType.TERMINATE: 'terminate'>
FAIL = <StopType.FAIL: 'fail'>
class Timer:

Create durable Timer conditions for Step waits.

Timer conditions measure durable workflow time rather than blocking a Python worker thread. Use a stable condition_id when external code may skip the Timer by ID.

@staticmethod
def by_duration( duration: datetime.timedelta, *, condition_id: str | None = None) -> dex.condition.Condition:

Create a Timer that becomes ready after duration.

Arguments:
  • duration: A non-negative delay measured from the wait's start.
  • condition_id: Optional stable identifier unique within the Wait tree.
Returns:

A durable Timer condition accepted by Wait.

Raises:
  • ValueError: If duration is negative or the identifier is invalid.
Examples:
>>> reminder = Timer.by_duration(timedelta(minutes=15), condition_id="remind")
>>> wait = Wait.until(reminder)
@dataclass(frozen=True)
class TimerId:

Select a Timer condition within a Step execution.

Exactly one selector must be set. Prefer condition_id for definitions that can evolve; condition_index is zero-based and follows flattened Wait order.

Attributes:
  • condition_id: A stable Timer condition identifier, or None.
  • condition_index: A zero-based Timer condition index, or None.
TimerId(condition_id: str | None = None, condition_index: int | None = None)
condition_id: str | None = None
condition_index: int | None = None
@staticmethod
def by_condition_id(condition_id: str) -> TimerId:

Select a Timer by its stable condition ID.

Arguments:
Returns:

A TimerId using the ID selector.

@staticmethod
def by_condition_index(condition_index: int) -> TimerId:

Select a Timer by zero-based Wait-tree position.

Arguments:
  • condition_index: The non-negative flattened Timer index.
Returns:

A TimerId using the index selector.

Raises:
@dataclass(frozen=True)
class Value:

Contain one validated SDK value before protocol mapping.

Attributes:
  • kind: The wire representation used for data.
  • data: The primitive value, bytes, or JSON text payload compatible with kind.
Value(kind: WireKind, data: str | bool | int | float | bytes)
kind: WireKind
data: str | bool | int | float | bytes
class ValueMappingError(builtins.RuntimeError):

Report an application-value encoding or decoding failure.

Attributes:
  • operation: The mapping operation, such as "encode" or "decode".
  • detail: The incompatible type, wire kind, or malformed payload detail.
ValueMappingError(operation: str, detail: str)

Create a mapping error with stable operation context.

Arguments:
  • operation: The attempted mapping operation.
  • detail: The human-readable failure detail.
operation
detail
@dataclass(frozen=True)
class Wait:

Describe the durable conditions evaluated before a Step executes.

A Step returns a Wait from wait_for. Dex persists the condition tree and invokes execute only when the selected readiness rule succeeds.

Attributes:
  • kind: The combination rule applied by Dex.
  • conditions: Direct Conditions for all-of or any-of waits.
  • combinations: Alternative all-of groups for any-combination waits.
Examples:
>>> wait = Wait.any_of(
...     replies.for_one(),
...     Timer.by_duration(timedelta(hours=1)),
... )
Wait( kind: dex.wait.WaitKind, conditions: tuple[dex.condition.Condition, ...] = (), combinations: tuple[ConditionCombination, ...] = ())
kind: dex.wait.WaitKind
conditions: tuple[dex.condition.Condition, ...] = ()
combinations: tuple[ConditionCombination, ...] = ()
@staticmethod
def skip_immediately() -> Wait:

Create a Wait that proceeds to execute immediately.

Returns:

A Wait with no Conditions.

@staticmethod
def until(condition: dex.condition.Condition) -> Wait:

Create a Wait for one Condition.

The Condition does not need a condition ID.

Arguments:
  • condition: The only readiness condition.
Returns:

An all-of Wait with condition.

@staticmethod
def all_of(*conditions: dex.condition.Condition) -> Wait:

Create a Wait that requires every Condition.

This combinator does not require condition IDs. Every Condition in any_combination_of() does require one.

Arguments:
  • *conditions: Conditions evaluated as one all-of group.
Returns:

An all-of Wait with Conditions in argument order.

@staticmethod
def any_of(*conditions: dex.condition.Condition) -> Wait:

Create a Wait that continues when any Condition is ready.

Conditions do not need condition IDs. Read Channel results from the Channel definition and inspect Timer outcomes through Context.

Among ready Conditions, Dex selects the first in canonical order: Timers, Channels, then SubFlows, preserving argument order within each kind. An unready Condition does not block a later ready Condition. To wait for strict priority, return only the higher-priority Condition.

Channel consumption is not greedy across alternatives. Dex consumes messages only from the selected Channel Condition; other ready Channel Conditions consume nothing.

Arguments:
  • *conditions: Alternative readiness Conditions.
Returns:

An any-of Wait containing the alternative Conditions.

@staticmethod
def any_combination_of(*combinations: ConditionCombination) -> Wait:

Create a Wait that accepts any complete Condition combination.

Channel consumption is not greedy across combinations. Dex consumes messages only from Channel Conditions in the selected combination; Conditions belonging only to other ready combinations consume nothing.

Arguments:
Returns:

A Wait containing the alternative groups.

class WaitForFailurePolicy(enum.Enum):

Control what Dex does when a Step's wait_for method fails.

Attributes:
  • FAIL_FLOW: Fail the Flow without invoking execute.
  • PROCEED: Invoke execute and expose the failure through Context.
FAIL_FLOW = <WaitForFailurePolicy.FAIL_FLOW: 'fail_flow'>
PROCEED = <WaitForFailurePolicy.PROCEED: 'proceed'>
class WireKind(enum.Enum):

Identify the protocol representation carried by an encoded Value.

Attributes:
  • STRING: A UTF-8 string value.
  • BOOL: A Boolean value.
  • INT64: A signed 64-bit integer value.
  • DOUBLE: A finite IEEE-754 double value.
  • BYTES: An uninterpreted byte sequence.
  • JSON: A canonical JSON document stored as text.
STRING = <WireKind.STRING: 'string'>
BOOL = <WireKind.BOOL: 'bool'>
INT64 = <WireKind.INT64: 'int64'>
DOUBLE = <WireKind.DOUBLE: 'double'>
BYTES = <WireKind.BYTES: 'bytes'>
JSON = <WireKind.JSON: 'json'>
class Worker:

Host synchronous Step and RPC handlers through outbound pull streams.

A Worker is one-shot. Call start on a dedicated thread because it blocks, then call stop or close during shutdown. Workers need only an outbound connection to the Dex Server. Handlers run concurrently in a bounded thread pool and must synchronize shared application state.

Attributes:
  • registry: The Flow Registry served by this Worker.
  • blob_cache: The shared cache used to hydrate large values.
  • options: The effective WorkerOptions.
Examples:
>>> worker = Worker(registry, cache)
>>> thread = threading.Thread(target=worker.start, daemon=True)
>>> thread.start()
>>> worker.wait_pull_registered(timeout=30)
>>> worker.stop()
Worker( registry: Registry, blob_cache: BlobCache, options: WorkerOptions | None = None)

Construct a Worker without starting its outbound pull streams.

Arguments:
  • registry: A Registry created for synchronous handlers.
  • blob_cache: An open cache shared for value hydration.
  • options: Networking, startup, and bearer credential options; None uses defaults and sends no authorization metadata.
Raises:
  • ValueError: If the pull Worker ID is invalid.
registry
blob_cache
options
def start(self) -> None:

Negotiate protocol, synchronize Attribute indexes, and serve.

start may be called exactly once. It first contacts FlowService using the configured synchronization timeout, then starts outbound pull streams and waits until stop finishes. Permanent stream rejections raise PullRegistrationError when stop_on_permanent_stream_rejection is enabled; otherwise streams retry.

Raises:
  • PullRegistrationError: A permanent stream rejection stops this Worker.
  • RuntimeError: If lifecycle state is invalid, Server protocol 4 is unsupported, concurrency exceeds matching.maxConcurrentTasksCap, or index synchronization fails.
def wait_pull_registered(self, timeout: float | None = None) -> None:

Wait until the required execution slot kinds have registered.

Start the Worker separately before waiting, or start it concurrently. Readiness requires one STEP_AND_RPC stream and, when RPC reservation is configured, one RPC_ONLY stream. Successful registration latches after stop. Use pull_registration_state for subsequent readiness probes. A permanent rejection before readiness raises PullRegistrationError only when no other stream is registered for that slot kind. Startup failures wake waiters. When the first registration reports a Server per-Worker limit below max_concurrent_tasks, every waiter raises that limit error and the Worker never becomes ready. Multiple callers may wait independently, including before start.

Arguments:
  • timeout: Maximum seconds to wait; None waits indefinitely.
Raises:
  • PullRegistrationError: A slot kind is permanently rejected before readiness.
  • TimeoutError: The timeout expires.
  • RuntimeError: The Worker stops before readiness, or max_concurrent_tasks exceeds the Server's matching.maxConcurrentTasksCap.
  • BaseException: The original startup failure, if start fails.
Examples:
>>> start_thread = threading.Thread(target=worker.start)
>>> start_thread.start()
>>> worker.wait_pull_registered(timeout=30)
>>> is_ready = worker.pull_registration_state().is_ready
pull_reserved_rpc_declines: int

Return the deprecated reserved RPC decline counter, always zero.

Reserved RPC capacity uses RPC_ONLY slots. Steps are never declined for capacity. This property remains available for source compatibility and is safe to read concurrently, including after stop.

Returns:

Zero.

Examples:
>>> declines = worker.pull_reserved_rpc_declines
def pull_registration_state(self) -> PullRegistrationState:

Return an immutable snapshot of execution slot kind registrations.

Transient failures, reconnects, and stop preserve initial readiness. A permanent rejection makes readiness false only while that slot kind has no registered stream. Re-registration clears its error. Calls are safe concurrently with start, stop, and other waiters. See wait_pull_registered for an example.

Returns:

Current state grouped by STEP_AND_RPC and, when configured, RPC_ONLY.

def stop(self) -> None:

Drain handlers and release Worker resources.

Calls before start and repeated calls after shutdown are safe. Shutdown drains for PullWorkerOptions.stop_drain_timeout (30 seconds by default), then requests cancellation and waits for every handler to return. A synchronous handler must check Context.is_cancellation_requested to respond promptly to forced shutdown. Call stop from a thread other than the pull Worker's start thread. Idle streams half-close immediately. Busy streams send final output before half-closing. Tasks received after stop begins are dropped without running handlers. A timed-out busy stream is reset before cancellation. Half-closed streams keep reading server frames until EOF or the remaining drain deadline before the channel closes. When the deadline expires, every stream that is still open is reset, even one blocked on a send or without Server EOF, so stop returns once handlers return. The Server may count a task it sent in that window as an attempt that may have run. One exception applies to a stream whose final output was queued or written but not yet confirmed by Server EOF or by the next assignment. Such a stream is not reset at the drain deadline, even if its handler finished or it already half-closed. It gets up to 5 more seconds to send the output, half-close, and receive Server EOF, and is reset after that. Stop can therefore return up to 5 seconds after the drain deadline.

Raises:
  • RuntimeError: If called on the pull Worker's start thread.
def close(self) -> None:

Stop the Worker and permanently mark it closed.

Uses the same shutdown contract as stop: idle streams half-close, busy streams send final output before half-closing, and later tasks are dropped. A handler exceeding the drain timeout has its stream reset. Pull streams finish their bounded EOF wait before the channel closes; a stream still open at the deadline is reset. close is idempotent; a closed Worker cannot be started again.

class WorkerInvocationError(dex.DexServiceError):

Expose both FlowService and nested Worker handler failure details.

Attributes:
  • code: The outer FlowService gRPC status.
  • sub_status: Usually WORKER_API_ERROR.
  • detail: The outer service detail.
  • operation: The Client operation that invoked the Worker.
  • flow_id: The targeted Flow ID, if available.
  • worker_code: The nested Worker gRPC status, or None when unavailable.
  • worker_error_type: The Worker-reported application error type.
  • worker_error_detail: The Worker-reported human-readable detail.
WorkerInvocationError( code: grpc.StatusCode, sub_status: ErrorSubStatus, detail: str, operation: str, flow_id: str | None, worker_code: grpc.StatusCode | None, worker_error_type: str, worker_error_detail: str)

Create an error from outer and nested Worker failure metadata.

Arguments:
  • code: The outer FlowService gRPC status.
  • sub_status: The Dex-specific classification.
  • detail: The outer service detail.
  • operation: The Client operation name.
  • flow_id: The targeted Flow ID, if any.
  • worker_code: The nested Worker gRPC status, if available.
  • worker_error_type: The nested application error type.
  • worker_error_detail: The nested human-readable detail.
worker_code
worker_error_type
worker_error_detail
@dataclass(frozen=True)
class WorkerOptions:

Configure Worker networking, pull polling, and startup synchronization.

attribute_index_sync_timeout is the RPC deadline used before the Worker opens its outbound task streams. It defaults to two minutes and must be positive.

Attributes:
  • server_address: The Dex FlowService target used for startup synchronization and blob hydration. Defaults to "localhost:8801".
  • attribute_index_sync_timeout: Maximum duration for startup Attribute index synchronization. Defaults to two minutes and must be positive.
  • pull: Outbound polling options. Defaults to PullWorkerOptions(). None is invalid because push Workers are no longer supported. Workers connect to server_address without an inbound listener.
  • credential: Supplies the current bearer token for FlowService calls and each pull stream open. None sends no authorization metadata. Token errors fail calls before Dex is contacted. Protect the plaintext network path. A rotated token is used on the next call or reconnect. AsyncWorker accepts awaitable tokens and bounds pull stream token lookup to 30 seconds. Synchronous token lookup must return promptly.
WorkerOptions( server_address: str = 'localhost:8801', attribute_index_sync_timeout: datetime.timedelta = datetime.timedelta(seconds=120), pull: PullWorkerOptions = <factory>, credential: Credential | None = None)
server_address: str = 'localhost:8801'
attribute_index_sync_timeout: datetime.timedelta = datetime.timedelta(seconds=120)
credential: Credential | None = None
@dataclass(frozen=True)
class PullWorkerOptions:

Configure outbound task polling for a Worker or AsyncWorker.

Pass this to WorkerOptions(pull=PullWorkerOptions()). Each stream runs one task at a time. Handlers may run concurrently and must synchronize shared state.

Attributes:
  • max_concurrent_tasks: Execution slots, each owning one long-lived poll stream. Zero uses the CPU count. Negative values are invalid. Start fails if this exceeds the Server's nonzero per-Worker limit reported by matching.maxConcurrentTasksCap.
  • reserved_rpc_tasks: Slots accepting only RPCs. The remaining slots accept Steps and RPCs. Steps are never declined for capacity. None reserves one slot when the resolved task limit is at least two, otherwise zero. Zero disables reservation. Values must be nonnegative and smaller than the resolved task limit. With one slot and no explicit reserve, start warns that RPCs may wait behind long-running Steps.
  • idle_streams_per_queue: Deprecated; accepted for source compatibility and has no effect. The Worker opens exactly max_concurrent_tasks streams. Zero remains valid; negative values remain invalid.
  • app_worker_id: Unique process identity for sticky routing. Empty generates an identity with eight random base32 characters. A supplied value must contain 1-63 ASCII letters, digits, dots, underscores, or hyphens, differ from every registered Flow type, and change after restart.
  • stop_drain_timeout: Grace period for active tasks on stop. Zero uses 30 seconds. Idle streams half-close immediately. Busy streams send final output before half-closing. Tasks arriving after stop begins are dropped. On timeout a busy stream is reset, its handler is canceled, and stop waits for the handler to return. Negative values are invalid. Busy Reconnect waits this long for server EOF after the final frame. During Stop, half-closed streams use the remaining drain deadline.
  • stop_on_permanent_stream_rejection: Whether the first permanent rejection leaving a slot kind without a registered stream drains the Worker and raises PullRegistrationError from start. Defaults to False, which retries indefinitely. Applies before and after initial readiness.
Raises:
  • ValueError: If a numeric option is negative, the reserve reaches the task limit, or app_worker_id is invalid.
Examples:
>>> pull = PullWorkerOptions(max_concurrent_tasks=8, reserved_rpc_tasks=2)
>>> worker = Worker(registry, cache, WorkerOptions(pull=pull))
PullWorkerOptions( max_concurrent_tasks: int = 0, idle_streams_per_queue: int = 0, app_worker_id: str = '', stop_drain_timeout: datetime.timedelta = datetime.timedelta(0), stop_on_permanent_stream_rejection: bool = False, reserved_rpc_tasks: int | None = None)
max_concurrent_tasks: int = 0
idle_streams_per_queue: int = 0
app_worker_id: str = ''
stop_drain_timeout: datetime.timedelta = datetime.timedelta(0)
stop_on_permanent_stream_rejection: bool = False
reserved_rpc_tasks: int | None = None
class PullRegistrationError(builtins.RuntimeError):

Identify a permanent rejection leaving a slot kind without a registered stream.

Raised by wait_pull_registered before initial readiness, or by start when stop_on_permanent_stream_rejection is enabled. Inspect queue and error for the rejected slot kind and original gRPC error. Permanent codes are INVALID_ARGUMENT, RESOURCE_EXHAUSTED, UNIMPLEMENTED, PERMISSION_DENIED, and UNAUTHENTICATED.

Attributes:
  • queue: Rejected slot kind: STEP_AND_RPC or RPC_ONLY. The field name is retained for source compatibility.
  • error: Original permanent transport error, also exposed as __cause__.
PullRegistrationError(queue: str, error: BaseException)

Create a slot kind registration rejection.

Arguments:
  • queue: Rejected slot kind label, retained in the queue field.
  • error: Original permanent transport error.
queue
error
@dataclass(frozen=True)
class PullRegistrationState:

Describe pull registration readiness independently of the Worker lifecycle.

Obtain snapshots with worker.pull_registration_state(). After every required slot kind has registered once, transient failures, reconnects, and stop keep readiness. A permanent error with no registered stream makes is_ready false until re-registration. When the first registration reports a Server per-Worker limit below max_concurrent_tasks, is_ready stays false permanently.

Attributes:
  • is_ready: Whether STEP_AND_RPC and, when reserved, RPC_ONLY registered once, the first registration passed the Server per-Worker limit check, and no required slot kind is permanently rejected without a registered stream.
  • queues: Immutable snapshots ordered STEP_AND_RPC then RPC_ONLY when reserved.
Examples:
>>> state = worker.pull_registration_state()
>>> is_ready = state.is_ready
PullRegistrationState( is_ready: bool = False, queues: tuple[PullQueueRegistrationState, ...] = ())
is_ready: bool = False
queues: tuple[PullQueueRegistrationState, ...] = ()
@dataclass(frozen=True)
class PullQueueRegistrationState:

Describe one execution slot kind in an immutable readiness snapshot.

Attributes:
  • queue: STEP_AND_RPC or RPC_ONLY; the field name is retained.
  • has_registered_stream: Whether a stream is currently registered.
  • last_permanent_error: Last permanent transport error, or None. A new Registered acknowledgement clears it. Treat errors as immutable.
PullQueueRegistrationState( queue: str, has_registered_stream: bool, last_permanent_error: BaseException | None)
queue: str
has_registered_stream: bool
last_permanent_error: BaseException | None
@dataclass(frozen=True)
class WaitForAttributeOptions:

Configure one durable Attribute match wait.

The server derives a stable Request ID from the Attribute condition when request_id is empty. Reuse an override only for the same logical predicate. request_timeout bounds the entire SDK call across transparent transport reattachments. Temporal permits 10 in-flight Updates per Workflow Execution. internal_handler_timeout reclaims accepted waits that outlive callers and could consume those slots. Active callers transparently start another generation, which adds another Update to history.

Attributes:
  • request_id: An optional override for the server-derived stable ID.
  • request_timeout: The total SDK call budget. Zero waits indefinitely.
  • internal_handler_timeout: The Temporal Update handler generation lifetime. Set it only when abandoned waits can approach Temporal's in-flight Update limit. Zero disables rollover. Prefer a value longer than normal request timeouts and reconnect gaps because every generation counts toward Temporal's 2,000-Update history limit.
WaitForAttributeOptions( request_id: str = '', request_timeout: datetime.timedelta = datetime.timedelta(0), internal_handler_timeout: datetime.timedelta = datetime.timedelta(0))
request_id: str = ''
request_timeout: datetime.timedelta = datetime.timedelta(0)
internal_handler_timeout: datetime.timedelta = datetime.timedelta(0)
@dataclass(frozen=True)
class WaitForStepCompletionOptions:

Configure one durable Step completion wait.

The server derives a stable Request ID from the Step execution when request_id is empty. Reuse an override only for the same logical wait. request_timeout bounds the entire SDK call across transparent transport reattachments. Temporal permits 10 in-flight Updates per Workflow Execution. internal_handler_timeout reclaims accepted waits that outlive callers and could consume those slots. Active callers transparently start another generation, which adds another Update to history.

Attributes:
  • request_id: An optional override for the server-derived stable ID.
  • request_timeout: The total SDK call budget. Zero waits indefinitely.
  • internal_handler_timeout: The Temporal Update handler generation lifetime. Set it only when abandoned waits can approach Temporal's in-flight Update limit. Zero disables rollover. Prefer a value longer than normal request timeouts and reconnect gaps because every generation counts toward Temporal's 2,000-Update history limit.
WaitForStepCompletionOptions( request_id: str = '', request_timeout: datetime.timedelta = datetime.timedelta(0), internal_handler_timeout: datetime.timedelta = datetime.timedelta(0))
request_id: str = ''
request_timeout: datetime.timedelta = datetime.timedelta(0)
internal_handler_timeout: datetime.timedelta = datetime.timedelta(0)
def dead_end() -> StepDecision:

End this execution path without scheduling work or closing the Flow.

Returns:

A dead-end decision; other active paths may continue.

def force_complete(output: object = <object object>) -> StepDecision:

Request immediate successful completion of the Flow.

Arguments:
  • output: Optional codec-supported Flow result. Omit it for no output.
Returns:

A force-completion decision that does not await other active Steps.

def force_complete_if_channels_empty( output: object, fallback: StepMovement[typing.Any], *channels: Union[Channel[Any], ChannelMap[Any]]) -> StepDecision:

Complete only when selected Channels are empty, otherwise schedule fallback.

Arguments:
  • output: The codec-supported completion result.
  • fallback: Movement scheduled when any selected Channel is non-empty.
  • *channels: Registered Channel or ChannelMap definitions to inspect.
Returns:

A conditional force-completion decision.

def force_fail(reason: str) -> StepDecision:

Request immediate Flow failure with an application reason.

Arguments:
  • reason: Non-empty human-readable failure detail.
Returns:

A force-failure decision.

def graceful_complete(output: object = <object object>) -> StepDecision:

Request successful completion after already-scheduled Steps finish.

Arguments:
  • output: Optional codec-supported Flow result. Omit it for no output.
Returns:

A graceful-completion decision.

def heartbeat(value: object = <object object>) -> StepOutput:

Create a heartbeat frame for a synchronous Step generator.

Yield heartbeat() to clear persisted heartbeat details. Passing a value, including None, persists that codec-supported value for the next regular activity attempt. Local activity execution ignores heartbeat frames.

Arguments:
  • value: Optional checkpoint value. Omitting it clears prior details.
Returns:

A StepOutput that must be yielded by the current synchronous handler.

Examples:
>>> def execute(context: Context, input: int) -> Generator[StepOutput, None, StepDecision]:
...     yield heartbeat({"completed": input})
...     return graceful_complete()
def go_to( step: Union[type[Step[~InputT]], Step[~InputT], StepRef], input: ~InputT) -> StepDecision:

Create a decision that schedules one next Step.

The Worker validates the target when it applies the returned decision. An unknown StepRef, ambiguous class, or unregistered instance raises FlowDefinitionError at that point, not when this function is called.

Arguments:
  • step: Destination class, registered instance, or StepRef. A class must be unambiguous.
  • input: A value compatible with the destination input annotation.
Returns:

A next decision containing one StepMovement.

def go_to_many(*movements: StepMovement[typing.Any]) -> StepDecision:

Create a decision that schedules several next Steps.

The Worker validates each movement target when it applies the returned decision, as described for go_to.

Arguments:
  • *movements: Typed destination movements applied in argument order.
Returns:

A next decision with all movements.

def open_blob_cache(config: BlobCacheConfig) -> BlobCache:

Open or create a native BlobCache.

Arguments:
  • config: Valid directory, byte capacity, and admission-policy settings.
Returns:

An open cache owned by the caller; call BlobCache.close() at shutdown.

Raises:
  • ValueError: If a configuration value is invalid.
  • OSError: If the directory cannot be created, locked, or opened.
Examples:
>>> cache = open_blob_cache(BlobCacheConfig(".dex-cache", 64 * 1024**2))
>>> try:
...     cache.put("blob-1", b"payload")
... finally:
...     cache.close()
def rpc( handler: Optional[~CallableT] = None, *, name: str | None = None, timeout: datetime.timedelta | None = None, lock_attributes: Sequence[AttributeLock] = (), is_transactional: bool = False, load_attribute_maps: Sequence[AttributeMap[Any]] = (), load_attribute_map_instances: Sequence[AttributeMapLoad] = (), load_channels: Sequence[Channel[Any]] = (), load_channel_maps: Sequence[ChannelMap[Any]] = (), load_channel_map_instances: Sequence[ChannelMapLoad] = ()) -> Union[~CallableT, Callable[[~CallableT], ~CallableT]]:

Mark a Flow method as a typed RPC handler.

Use @rpc with defaults or call @rpc(...) with configuration. A handler receives Context plus zero or one typed input and returns RPCResult[T] or None. Registry construction validates annotations and Attribute locks.

Arguments:
  • handler: The decorated bound-method function when used as @rpc.
  • name: Optional protocol RPC name; defaults to the method name.
  • timeout: Optional non-negative handler timeout.
  • lock_attributes: Attribute locks held for the entire handler invocation.
  • is_transactional: Request transactional reads and writes even without locks.
  • load_attribute_maps: AttributeMaps whose current instances are loaded.
  • load_attribute_map_instances: Exact instances created by AttributeMap.load.
  • load_channels: Channels whose pending messages are loaded.
  • load_channel_maps: ChannelMaps whose current instance messages are loaded.
  • load_channel_map_instances: Exact instances created by ChannelMap.load_messages.
Returns:

The original handler, or a decorator that returns it after attaching options.

Raises:
  • ValueError: If name is empty or timeout is negative.
Examples:
>>> class Orders(Flow[None]):
...     @rpc(timeout=timedelta(seconds=10), lock_attributes=(status.lock(),))
...     def cancel(self, context: Context, reason: str) -> RPCResult[bool]:
...         status.set(context, "canceled")
...         return RPCResult(True)
def retry_after( after_seconds: int, cause: BaseException) -> RetryAfterError:

Creates an error that sets the delay before Dex retries a worker method.

Raise the returned error from a Step method after an external dependency reports that it is not ready. For example::

raise retry_after(5, RuntimeError("service is not ready"))
Arguments:
  • after_seconds (int): Positive delay in seconds before the next retry.
  • cause (BaseException): Failure that Dex records as the worker error.
Returns:

RetryAfterError: Error to raise from the worker method.

Raises:
  • ValueError: If after_seconds is not positive or cause is None.