_types

Single keypress.

Event dispatched to a reducer.

Named screen config: template name + reducer reference.

Flow edge between screens.

Declarative field configuration.

Runtime state for a single field.

Full form state.

Call a function, resume saga with its return value.

Dispatch an action back to the store.

Read current state, resume saga with it.

Run another saga concurrently.

By default forks are detached: the child gets its own cancellation scope and is not cancelled when the parent is. Setattached=True to inherit the parent's cancellation scope — cancelling the parent will transitively cancel the child.

Sleep for N seconds.

Call a function with retry and backoff on failure.

Usage in a saga::

result = yield Retry(fetch_data, args=(url,), max_attempts=3, backoff="exponential")

Wrap a blocking effect with a deadline.

Usage in a saga::

result = yield Timeout(Call(fetch_data, args=(url,)), seconds=5)

RaisesTimeoutErrorif the effect doesn't complete in time. Only wraps blocking effects (Call and Retry).

Call a function, returning (result, None) on success or (None, error) on failure.

UnlikeCall, exceptions do not crash the saga::

result, error = yield TryCall(fn=might_fail)
if error:
    yield Put(Action("FETCH_FAILED", payload=str(error)))
else:
    yield Put(Action("FETCH_OK", payload=result))

Run multiple sagas concurrently, return the first result.

Losers are cancelled via their cancel events as soon as a winner completes. If all racers fail, the first error is thrown into the parent saga::

winner = yield Race(sagas=(fetch_primary(), fetch_fallback()))

RaisesStateErrorif sagas is empty.

Run multiple sagas concurrently, wait for all to complete.

Returns a tuple of results in the same order as the input sagas. Fail-fast: if any saga raises, remaining sagas are cancelled and the error is thrown into the parent::

a, b = yield All(sagas=(fetch_users(), fetch_roles()))

An empty tuple returns()immediately.

Pause the saga until a matching action is dispatched.

Waits for future actions only — actions dispatched before the Take is yielded are not matched. Returns the fullAction object so the saga can inspect both type and payload::

action = yield Take("USER_CONFIRMED")
name = action.payload["name"]

An optional timeout (in seconds) raisesTimeoutErrorif the action is not dispatched in time.

Delay-then-fork: start a timer, fork saga when it expires.

If the parent saga yields anotherDebouncebefore the timer fires, the previous timer is cancelled and restarted. The parent continues immediately (non-blocking)::

# In a keystroke handler saga:
while True:
    key = yield Take("@@KEY")
    yield Debounce(seconds=0.3, saga=search_saga)

The debounced saga runs independently; useTakeif the parent needs the result.

Fork a new saga for every matching action (auto-restart pattern).

Blocks the parent saga until cancelled. For each dispatched action whose type matches action_type, a new saga is forked with the action as argument::

# Fork a handler for every CLICK action:
yield TakeEvery("CLICK", handle_click)

def handle_click(action):
    url = action.payload["url"]
    result = yield Call(fetch, args=(url,))
    yield Put(Action("FETCHED", payload=result))

The watcher loop runs until the parent saga is cancelled.

Fork a saga for the latest matching action, cancelling the previous.

LikeTakeEverybut only the most recent saga runs — when a new matching action arrives, the previous fork is cancelled before the new one starts::

# Only the latest search runs:
yield TakeLatest("SEARCH", run_search)

Useful for typeahead/autocomplete patterns where earlier results are obsolete.

Lightweight command: a thunk that returns an Action (or None).

Simpler than sagas for one-shot effects::

def fetch_cmd():
    data = fetch_json(url)
    return Action("FETCH_DONE", payload=data)

return ReducerResult(state, cmds=(Cmd(fetch_cmd),))

Run commands concurrently with no ordering guarantees.

Usage::

return ReducerResult(state, cmds=(Batch(cmd_a, cmd_b, cmd_c),))

Run commands serially, in order.

Each command's result is dispatched before the next starts::

return ReducerResult(state, cmds=(Sequence(cmd_a, cmd_b),))

Schedule a single @@TICK after interval seconds.

Return from a reducer to start ticking. Return another TickCmd when you receive @@TICK to keep the loop going; omit to stop::

case "@@TICK":
    if state.loading:
        return ReducerResult(new_state, cmds=(TickCmd(0.15),))
    return new_state

Declarative terminal state returned alongside rendered content.

The renderer diffs previous vs. current ViewState and applies only the changes. Attach to ReducerResult to control terminal features from your reducer::

return ReducerResult(state, view=ViewState(cursor_visible=True))

Reducer can return this to trigger side effects.

Signal the app to exit. Return from a reducer to stop the event loop.

Strip None entries and simplify command tuples.

Returns an empty tuple for no commands, a single-element tuple for one, and the full tuple otherwise. Avoids unnecessary allocations for the common zero-or-one-command case.