Module

state

Store, dispatch, saga runner, combine_reducers.

1Class3Functions

Classes

Store 16 ▼
Centralized state container with saga support. Thread-safety: reads are lock-free (frozen state). …

Centralized state container with saga support.

Thread-safety: reads are lock-free (frozen state). Dispatch serializes through a lock. Sagas run on a ThreadPoolExecutor.

Methods

state 0 Any ▼
property
def state(self) -> Any
Returns
Any
view_state 0 Any ▼
Latest ViewState from a reducer, or None.
property
def view_state(self) -> Any
Returns
Any
quit_requested 0 bool ▼
True if a reducer returned Quit.
property
def quit_requested(self) -> bool
Returns
bool
exit_code 0 int ▼
Exit code from the Quit signal (default 0).
property
def exit_code(self) -> int
Returns
int
recording 0 list[dict] | None ▼
Get session recording if enabled.
property
def recording(self) -> list[dict] | None
Returns
list[dict] | None
dispatch 1 ▼
Dispatch action through middleware -> reducer.
def dispatch(self, action: Action) -> None
Parameters
Name Type Description
action —
run_saga 1 ▼
Schedule a saga on the thread pool.
def run_saga(self, saga: Any) -> None
Parameters
Name Type Description
saga —
subscribe 1 Callable[[], None] ▼
Register state-change listener. Returns unsubscribe callable.
def subscribe(self, listener: Callable) -> Callable[[], None]
Parameters
Name Type Description
listener —
Returns
Callable[[], None]
shutdown 0 ▼
Shut down the thread pool, waiting for pending work.
def shutdown(self) -> None
Internal Methods 7 ▼
__init__ 4 ▼
def __init__(self, reducer: Callable, initial_state: Any, middleware: tuple[Callable, ...] = (), *, record: bool | str | Path = False) -> None
Parameters
Name Type Description
reducer —
initial_state —
middleware — Default:()
record — Default:False
_base_dispatch 1 ▼
Core dispatch: reducer + saga scheduling + cmd execution + recording.
def _base_dispatch(self, action: Action) -> None
Parameters
Name Type Description
action —
_run_saga 1 ▼
Step through a generator saga, executing effects. Catches unhandled exceptions…
def _run_saga(self, saga: Any) -> None

Step through a generator saga, executing effects.

Catches unhandled exceptions and dispatches @@SAGA_ERROR so the reducer can react gracefully. The error is never swallowed silently.

Parameters
Name Type Description
saga —
_exec_cmd 1 ▼
Execute a Cmd, Batch, Sequence, or TickCmd.
def _exec_cmd(self, cmd: Any) -> None
Parameters
Name Type Description
cmd —
_run_cmd 1 ▼
Run a single Cmd thunk and dispatch its result.
def _run_cmd(self, fn: Any) -> None
Parameters
Name Type Description
fn —
_run_sequence 1 ▼
Run commands serially, dispatching each result before the next.
def _run_sequence(self, cmds: tuple) -> None
Parameters
Name Type Description
cmds —
_run_tick 1 ▼
Schedule a single @@TICK after *interval* seconds.
def _run_tick(self, interval: float) -> None
Parameters
Name Type Description
interval —

Functions

combine_reducers 1 Callable ▼
Combine multiple reducers into one that manages a dict state. Each reducer man…
def combine_reducers(**reducers: Callable) -> Callable

Combine multiple reducers into one that manages a dict state.

Each reducer manages a slice of state under its keyword name. Sagas, cmds, and view state from ReducerResult and Quit are collected and propagated.

Parameters
Name Type Description
**reducers Callable
Returns
Callable
_merge_view 2 ViewState ▼
Merge two ViewStates: explicitly-set fields in *new* override *prev*.
def _merge_view(prev: ViewState | None, new: ViewState) -> ViewState
Parameters
Name Type Description
prev ViewState | None
new ViewState
Returns
ViewState
_execute_retry 7 Any ▼
Execute a function with retry and backoff.
def _execute_retry(fn: Any, args: tuple, kwargs: dict, max_attempts: int, backoff: str, base_delay: float, max_delay: float) -> Any
Parameters
Name Type Description
fn Any
args tuple
kwargs dict
max_attempts int
backoff str
base_delay float
max_delay float
Returns
Any