跳到主要内容

Client Advanced

等待 Flow​

WaitForFlow 等待 Flow 到达 terminal status。bounded wait 在 Flow 仍然 open 时会返回 long-poll timeout；再次调用即可继续等待。

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

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

等待 Step​

WaitForStepCompletion 在 Flow 继续运行时等待一个 StepExecutionID。它只确认完成，不返回 Step output。

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

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

等待 Attribute 匹配​

WaitForAttributeMatch 等待 Attribute 满足类型化的 scalar 比较，并返回当前命中的值。它定位当前 active run。对于 AttributeMap，还需要传入 instance。

String 和 Boolean 支持 equal 与 not-equal。整数和浮点数还支持 greater-than、greater-than-or-equal、less-than 与 less-than-or-equal。缺失的 Attribute 永远不命中。

当调用方需要在状态变化后刷新时，使用应用拥有的 revision Attribute。等待 revision 大于上次观察到的值，再调用 Describe 或 read RPC。Revision 是 watermark；快速变化可能合并为一个返回值。

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

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

object、bytes、null、blob-backed、非有限浮点、跨类型比较和无效 ordering 比较会被拒绝。该等待只支持 Temporal；Cadence 返回 Unimplemented。完整 revision 模式见等待 Attribute 匹配。

查看 Flow​

DescribeFlow 返回 current 或 latest run 的 IDs、Flow type、status 和 start time。不要求 Flow 仍然 active。

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

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

搜索 Flow​

SearchFlows 返回匹配 visibility query 的一页 Flow runs。search 用来寻找 Flow instance；DescribeFlow 用于查看一个已知 Flow ID。要继续搜索，传回 page token。

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

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

跳过 Timer​

SkipTimer 让一个等待中的 Timer 立刻 ready。选择 Step execution，以及 Timer 的 condition ID 或它从零开始的 condition index。

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

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

更新 FlowConfig​

UpdateFlowConfig 修改 FlowConfig。

await client.update_flow_config("order-123", FlowConfig(continue_as_new_threshold=1_000))

例子: examples/python/dex_examples/primitives/flow/controller.py

触发 continue-as-new​

TriggerContinueAsNew 请求 active Flow 把 history 滚动到新的 run。

await client.trigger_continue_as_new("order-123")

例子: examples/python/dex_examples/primitives/flow/controller.py

Time travel​

TimeTravel 从已有 Flow history 的某个点创建一个新 run。

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

例子: examples/python/dex_examples/primitives/flow/controller.py