Skip to main content

Client Advanced

Requests during Flow close​

While a Flow closes, Dex rejects new transactional or locking RPCs, WaitForStepCompletion, WaitForAttributeMatch, and Channel message deletion with gRPC FAILED_PRECONDITION and sub-status FLOW_CLOSING. Each SDK exposes a typed Flow closing error. Re-read the retained terminal Flow state before deciding whether any further action is needed. The error does not prove that an earlier request succeeded.

Wait for a Flow​

WaitForFlow waits for terminal Flow status. A bounded wait can return a long-poll timeout while the Flow is still open; call it again to keep waiting.

result = await client.wait_for_flow("order-123", timedelta(minutes=1))

Example: examples/python/dex_examples/primitives/client-apis/controller.py

Wait for a Step​

WaitForStepCompletion waits for one StepExecutionID while the Flow continues. It confirms completion but does not return the Step output.

await client.wait_for_step_completion("order-123", StepExecutionId("ChargeOrder"), timedelta(minutes=1))

Example: examples/python/dex_examples/products/order-processing/controller.py

Wait for an Attribute match​

WaitForAttributeMatch waits until an Attribute satisfies a typed scalar comparison and returns the matched current value. It targets the current active run. For an AttributeMap, also supply the instance.

String and Boolean values support equal and not-equal. Integer and floating-point values also support greater-than, greater-than-or-equal, less-than, and less-than-or-equal. A missing Attribute never matches.

Use an application-owned revision Attribute when callers need to refresh after state changes. Wait for a value greater than the last revision, then invoke a Describe or read RPC. Revision is a watermark: rapid changes may be coalesced into one returned value.

revision = await app_state.client.wait_for_attribute_match(
    flow_id,
    app_state.job_post.update_version,
    AttributeMatch.greater_than(required_int_query("lastRevision")),
    timedelta(seconds=30),
)

Example: examples/python/dex_examples/products/job-post/controller.py

Object, bytes, null, blob-backed, non-finite floating-point, cross-type, and invalid ordering comparisons are rejected. See Wait for Attribute Match for the complete revision pattern.

Describe a Flow​

DescribeFlow reads the current or latest run's IDs, Flow type, status, and start time. The Go SDK also returns the close time when available. It does not wait for the Flow to finish. Use it for a one-off status check after a read-only RPC returns a non-terminal business state. A missing ID or an ID that does not belong to a Flow returns a typed Flow-not-found error.

info = await client.describe_flow("order-123")

Example: examples/python/dex_examples/primitives/client-apis/controller.py

Search Flows​

SearchFlows returns one page of Flow runs matching a visibility query. Internal server executions are excluded. Search finds Flow instances; DescribeFlow describes one known Flow ID. Pass the returned page token to continue a search.

Keep the query unchanged across pages.

page = await client.search_flows("FlowType = 'ClientApisFlow'", 20, "")

Example: examples/python/dex_examples/primitives/client-apis/controller.py

Skip a Timer​

SkipTimer makes one waiting Timer ready immediately. Choose the Step execution and either the Timer's condition ID or its zero-based condition index.

await client.skip_timer("order-123", StepExecutionId("ChargeOrder"), TimerId.by_condition_id("retry"))

Example: examples/python/dex_examples/products/order-processing/controller.py

Update FlowConfig​

UpdateFlowConfig changes the FlowConfig of an active Flow. Send only the fields you want to change. The SDK update does not change the Flow's Worker routing. See Update FlowConfig with a Client for the fields.

Trigger continue-as-new​

TriggerContinueAsNew is deprecated. Dex does not continue Flows as new. The call always fails with UNIMPLEMENTED. Do not call it. A long-running Flow needs no rollover, because Dex Server keeps a recent history window.

Time travel​

TimeTravel creates a new run attempt from a point in an existing Flow's history. The run keeps its Run ID. The point must be inside the recent history window. Otherwise the call fails with FAILED_PRECONDITION. Time travel to the beginning is always available.

new_run_id = await client.time_travel("order-123", TimeTravelOptions(type=TimeTravelType.BEGINNING, reason="code fix"))

Example: examples/python/dex_examples/primitives/flow/controller.py