Cron schedule
Run fixed-interval work in one durable Flow.
Flow definition
The Flow has three Step definitions:
- InitStep validates a positive interval and run count, then moves to WaitForSchedule.
- WaitForSchedule waits for the interval Timer, Trigger, or Skip. Skip consumes an occurrence and starts the next interval. The timer or Trigger starts Run and the next WaitForSchedule occurrence together.
- Run does the scheduled work. It reaches a dead end unless this is the final occurrence, which completes the Flow.
External code uses the Client to publish the Trigger or Skip Channel.
Core implementation
Each sample shows the WaitForSchedule Step definition and its scheduling decision. The runnable example also includes input types, Flow registration, and the work implementation.
class _WaitForSchedule(Step[_ScheduleState]):
def __init__(
self,
trigger: Channel[None],
skip: Channel[None],
) -> None:
self.trigger = trigger
self.skip = skip
def wait_for(self, context: Context, state: _ScheduleState) -> Wait:
return Wait.any_of(
Timer.by_duration(state.interval.duration()),
self.trigger.for_one(),
self.skip.for_one(),
)
def execute(self, context: Context, state: _ScheduleState) -> StepDecision:
if self.skip.results(context):
return self._next_schedule(state)
run_input = _RunInput(
run_number=state.remaining_runs,
is_final=state.remaining_runs == 1,
)
if run_input.is_final:
return go_to(_Run, run_input)
return go_to_many(
StepMovement.of(_Run, run_input),
StepMovement.of(
_WaitForSchedule,
_ScheduleState(state.interval, state.remaining_runs - 1),
),
)
Example: examples/python/dex_examples/patterns/cron/cron_schedule_flow.py