wijjit.core.state.State

class wijjit.core.state.State(data=None)

Application state with change detection.

This class behaves like a dictionary but tracks changes and can trigger callbacks when values are modified. It also supports attribute-style access for convenience.

Parameters:

data (dict, optional) – Initial state data

Variables:

• data (dict) – The underlying state dictionary

• _change_callbacks (list) – List of callbacks to trigger on state changes

• _watchers (dict) – Dictionary mapping keys to their specific watchers

Examples

>>> state = State({'count': 0})
>>> state['count'] = 1  # Triggers change callback
>>> state.count = 2  # Also triggers change callback (attribute access)
>>> print(state.count)  # Access via attribute
2

__init__(data=None)

Parameters:

data (dict[str, Any] | None)

Return type:

None

Methods

__init__([data])
async_batch_update()	Async context manager for batch state updates.
batch_update()	Context manager for batch state updates.
clear()
copy()	Create a shallow copy of the state data, excluding callbacks.
flush_pending_async()	Wait for all pending async callback tasks to complete.
fromkeys(iterable[, value])
get(k[,d])
items()
keys()
off_change(callback)	Unregister a global state-change callback.
on_change(callback)	Register a callback for any state change.
pop(k[,d])	value.
popitem()	as a 2-tuple; but raise KeyError if D is empty.
reset([data])	Reset state to new data.
set_async(key, value)	Set a state value and await all triggered callbacks (async).
setdefault(k[,d])
unwatch(key[, callback])	Stop watching a state key.
update(other, /, **kwargs)	Update multiple state values at once.
values()
watch(key, callback)	Watch a specific state key for changes.

__deepcopy__(memo)

Create a deep copy of the state data only, excluding callbacks.

Parameters:

memo (dict) – Memoization dictionary for deepcopy

Returns:

A new State instance with deep-copied data but no callbacks

Return type:

State

Notes

This method is called by copy.deepcopy(). It copies only the state data, not the change callbacks or watchers. This is intentional because: 1. Callbacks are typically bound methods that reference the original app 2. The copied state is used for template rendering, not modification 3. Callbacks may contain unpicklable objects (thread locks, etc.)

copy()

Create a shallow copy of the state data, excluding callbacks.

Returns:

A new State instance with the same data but no callbacks or watchers.

Return type:

State

Notes

This override is required. UserDict.copy reassigns self.data on the instance, which State.__setattr__ rejects (data is the backing store and replacing it wholesale would fire no change callbacks). Inheriting it therefore raised StateKeyError on every call. Callbacks are dropped for the same reason as in __deepcopy__(): they are bound to the originating app.

__setitem__(key, value)

Set an item and trigger change callbacks.

Parameters:

• key (str) – The state key

• value (Any) – The new value

Return type:

None

Notes

Any key is allowed, including names that shadow a State or dict method (state["items"]). See _SHADOWED_NAMES for what that costs.

Change detection compares the new value against the live object already stored under key. That is exact for immutable values, but for a mutable container it means the pre-mutation snapshot is gone the moment the caller mutates it in place. Two consequences:

• state[k] = state[k] (the same object back again) cannot be shown to be unchanged, so it fires - see _is_aliased_mutable(). This is the supported escape hatch after an in-place mutation, and it warns, because building the new container first is better.

• state[k] = list(state[k]) after an in-place mutation is a new object that is value-equal to the already-mutated original, so it is indistinguishable from a genuine no-op write and stays silent. There is no fix short of snapshotting every container handed out on read.

The rule that always works: build the new container first, then assign (state[k] = [*state[k], x]). Order matters, and only the copy-then-mutate direction is detectable.

__getattr__(name)

Get state value via attribute access.

Parameters:

name (str) – The state key

Returns:

The state value

Return type:

Any

Raises:

AttributeError – If the key doesn’t exist

__setattr__(name, value)

Set state value via attribute access.

Parameters:

• name (str) – The state key

• value (Any) – The new value

Raises:

StateKeyError – If name shadows a State or dict method (see _SHADOWED_NAMES). The key itself is legal - use state[name] = value.

Return type:

None

on_change(callback)

Register a callback for any state change.

Supports both synchronous and asynchronous callbacks.

Parameters:

callback (callable or async callable) – Function to call when any state changes. Signature: callback(key, old_value, new_value) Can be sync or async.

Return type:

None

off_change(callback)

Unregister a global state-change callback.

The counterpart to on_change() (mirrors watch() / unwatch()). Removing a callback that was never registered is a no-op.

Parameters:

callback (callable or async callable) – The callback previously passed to on_change().

Return type:

None

watch(key, callback)

Watch a specific state key for changes.

Supports both synchronous and asynchronous callbacks.

Parameters:

• key (str) – The state key to watch

• callback (callable or async callable) – Function to call when this key changes. Signature: callback(key, old_value, new_value) Can be sync or async.

Return type:

None

unwatch(key, callback=None)

Stop watching a state key.

Parameters:

• key (str) – The state key to stop watching

• callback (callable, optional) – Specific callback to remove. If None, removes all watchers for this key.

Return type:

None

batch_update()

Context manager for batch state updates.

Use this to update multiple state values while suppressing intermediate callbacks. Callbacks are triggered once at the end of the batch for each key that actually changed.

Returns:

Context manager for batch updates

Return type:

_BatchContext

Examples

>>> state = State({'a': 1, 'b': 2})
>>> with state.batch_update():
...     state['a'] = 10
...     state['b'] = 20
...     state['a'] = 100  # Only this final value is used
# Callbacks triggered once for 'a' (1 -> 100) and once for 'b' (2 -> 20)

Notes

For async callbacks, this method schedules them as background tasks but cannot await their completion. Use async_batch_update() in async contexts for proper async callback handling.

async_batch_update()

Async context manager for batch state updates.

Use this in async contexts to update multiple state values while suppressing intermediate callbacks. All async callbacks are properly awaited when the context manager exits.

Returns:

Async context manager for batch updates

Return type:

_AsyncBatchContext

Examples

>>> state = State({'a': 1, 'b': 2})
>>> async with state.async_batch_update():
...     state['a'] = 10
...     state['b'] = 20
...     state['a'] = 100  # Only this final value is used
# Callbacks awaited for 'a' (1 -> 100) and 'b' (2 -> 20)

Notes

This method should be preferred over batch_update() when in an async context, as it ensures all async callbacks complete before continuing execution.

async set_async(key, value)

Set a state value and await all triggered callbacks (async).

This method provides guaranteed completion semantics for state changes. Unlike the sync __setitem__, this method awaits all callbacks before returning, ensuring state changes are fully processed.

Parameters:

• key (str) – The state key

• value (Any) – The new value

Raises:

ValueError – If key is a reserved dict method name

Return type:

None

Notes

Use this method when you need to ensure all state change side effects have completed before proceeding. Async callbacks are awaited, sync callbacks are run in the default executor to avoid blocking.

Examples

>>> state = State()
>>> async def save_to_db(key, old, new):
...     await db.save(key, new)
>>> state.on_change(save_to_db)
>>> await state.set_async("user", {"name": "Alice"})
# Database save is guaranteed to have completed

async flush_pending_async()

Wait for all pending async callback tasks to complete.

This method waits for any background tasks created by state changes (via __setitem__) to finish executing, including tasks created on the loop thread on behalf of a worker-thread state mutation (see _schedule_state_coroutine()) and tasks spawned by those callbacks while they run.

Notes

Useful for testing or ensuring cleanup before shutdown. In normal operation, you usually want fire-and-forget behavior for callbacks.

An initial await asyncio.sleep(0) gives a just-scheduled call_soon_threadsafe task-creator (worker-thread path) a chance to run before the set is even sampled, so its task is visible here. The loop then keeps gathering until the set is empty, since a callback can itself trigger further state changes whose tasks are added to _pending_tasks while this method is awaiting the current batch.

Examples

>>> state['count'] = 1  # Triggers async callback as background task
>>> await state.flush_pending_async()  # Wait for callback to finish

Return type:

None

update(other, /, **kwargs)

Update multiple state values at once.

Parameters:

• other (dict) – Dictionary of values to update

• **kwargs (Any) – Additional key-value pairs to update

Return type:

None

Notes

Each update triggers change callbacks via __setitem__.

reset(data=None)

Reset state to new data.

Parameters:

data (dict, optional) – New state data. If None, clears all state.

Return type:

None