Polling with Timer

PollingWithTimerFlow makes the wait explicit. PollingStep waits for a Timer, checks the external condition, then either completes or moves to itself for another interval. It uses more code than retry-based polling, but the not-ready outcome is a normal Step decision rather than a failed attempt.

Definition graph

PollingWithTimerFlow

Valid

python · examples/python/dex_examples/patterns/polling/simple_polling_flow.py

PollingWithTimerFlow

PollingStepSTART

WaitFor until

⏱️ POLLING_INTERVAL timer

?

gracefulComplete

goTo

🧩 PollingStep

Press enter or space to select a node. You can then use the arrow keys to move the node around. Press delete to remove it and escape to cancel.

Press enter or space to select an edge. You can then press delete to remove it or escape to cancel.

Core implementation

class PollingStep(Step[None]):
    def wait_for(self, context: Context, input: None) -> Wait:
        return Wait.until(Timer.by_duration(POLLING_INTERVAL))

    def execute(self, context: Context, input: None) -> StepDecision:
        if self._is_system_ready():
            return graceful_complete()
        return go_to(PollingStep, None)

Example: examples/python/dex_examples/patterns/polling/simple_polling_flow.py