A thread-safe AppState as an Extension of our "always-valid" Domain

Imagine a digital shopping cart where you can remove items and your total down in your UI component is adjusted accordingly. But the price in the top right corner of your website is only updated once you hit refresh - a typical app state management issue, a UI state issue to be more specific.

Further examples of app states would be if you are already logged in or if you should be forwarded to the login page - navigation can be part of our validatable AppState but is usually strongly tied to the frontend technology we use.

In our example project the previous version of the users page managed selection state with two one-element lists. selected held the currently highlighted user; _refresh_detail held a callable that the list card would invoke after mutating selected[0].

selected: list[User | None] = [None]
_refresh_detail: list[Callable[..., Any]] = [lambda: None]

def on_select(user: User) -> None:
    selected[0] = user
    _refresh_detail[0]()

Adding a second view component that cared about which user was selected would have required threading another holder through users_page. The more fundamental problem was that the state was invisible to the domain: anything could be written into selected[0], the update was only delivered to components that on_select explicitly knew about, and if a caller forgot to call the holder, views went stale silently.

Always-Valid State

The replacement is AppState, a frozen Pydantic model that also inherits from Record:

class AppState(BaseModel, Record):
    model_config = ConfigDict(frozen=True)
    selected_user: User | None = None

Pydantic's frozen=True gives AppState a consistent hash() based on its contents, which the concurrency mechanism depends on. Inheriting from Record provides .from_() and .with_() methods that run the full construction and validation pipeline and return a Result[AppState]. Any field_validator added to AppState later is therefore enforced at every write path automatically. The state cannot end up in a shape that violates its own invariants, for the same reason a User's Email with a missing @ cannot be constructed — the constraint lives in the class, not scattered across callers.

Invariants

When talking about "always-valid" domain objects, we have to mention invarints. Invariants are object constellations we want to enforce/restrict in our domain, sometimes our frameworks and tools already help us with that, think of an radio button component which ensures always exactly one option is selected. But sometimes the defaults of our programming language are not sufficient to enforce our business rules, like using a string to represent an email address, a valid string does not gurantee a valid email address -> further restrictions need to be enforced. Similarly a GUI application might allow to open overlay-views but only in certain menus and only one at a time. It is our job as developer to ensure the state of the app (which views and information is shown to the user) can be validated and to only permit display of valid constellations. Another example would be graph application, where you can add and delete datapoints - a statistics view shows you derived data, like averages - a user action deleting any data should result in the derived data to be updated accordingly, ideally within the same rendered frame, so a screenshot showing out-of-sync, or simply wrong data becomes impossible.

The Mediator Pipeline

Writes go through SetAppStateCommand, dispatched via Mediatr:

class SetAppStateCommand(BaseModel, GenericQuery, Record):
    model_config = ConfigDict(frozen=True)
    new_state: AppState
    last_state_hash: int

Extending GenericQuery means the existing log_performance_and_type behavior runs for every state write without any changes to that behavior. Three additional behaviors are registered specifically for SetAppStateCommand. app_state_mutex_behavior acquires an asyncio.Lock on the AppStateWrapper singleton before calling into the rest of the chain. app_state_modification_behavior, which runs inside that lock, checks the hash, updates the wrapper on success, and fires subscriber notifications. app_state_validation_behavior re-constructs the incoming state through AppState.from_() before the handler runs, so any validators added in the future are enforced at pipeline level rather than only at the call site.

This composition is the practical value of pipeline behaviors: each concern is written once and applied in a fixed order without the handler or the caller knowing it is there. When a new cross-cutting requirement appears — rate-limiting writes, audit logging — it slots in as another behavior rather than a change to existing code.

Hash Checks and Collision Handling

The last_state_hash in SetAppStateCommand is the hash of whatever state the caller read before constructing its update:

new_state_result = wrapper.current_state.with_(selected_user=user)
if has_errors(new_state_result): ...
cmd = SetAppStateCommand(
    new_state=new_state_result.value,
    last_state_hash=hash(wrapper.current_state),
)

current_state is a validate-with-resolute Record subtype, calling with_() performs an in-mutable, partial update on the last known state, ensures its validity via pydantic and packs it into the result type.
If a second coroutine writes a new state between that read and the dispatch, the hashes diverge and app_state_modification_behavior short-circuits. The caller receives a Result carrying a HashOutdatedError rather than silently overwriting a change it never saw. user_list_card handles this by retrying up to three times with a 50ms back-off, re-reading the fresh hash on each attempt, and distinguishing the retryable case from non-retryable errors by type:

if outcome.remove_errors_of_type([HashOutdatedError]).has_errors:
    return  # not a hash conflict, do not retry
await asyncio.sleep(0.05)

Result as the State Modification Contract

Every SetAppStateCommand dispatch returns a Result[AppState]. Hash conflicts, validation failures, and success all travel the same return path. The caller cannot ignore the outcome — the result has to be inspected before the value is accessible — and it can distinguish failure modes by type without parsing message strings.

The test suite demonstrates this clearly. result.contains_error_type(HashOutdatedError) verifies that a concurrent write with a stale hash failed for exactly the right reason. remove_errors_of_type filters out the retryable case from permanent failures. Both assertions would require string matching or isinstance checks scattered across exception handlers in a raise-based design.

Subscriber Delivery and View Consistency

AppStateWrapper holds the subscriber list. user_detail_card registers a single callback that triggers _content.refresh() whenever the wrapper notifies:

on_state_change = lambda _: _content.refresh()
wrapper.subscribe(on_state_change)
ui.context.client.on_disconnect(lambda: wrapper.unsubscribe(on_state_change))

The @ui.refreshable inner function reads wrapper.current_state.selected_user on each render, so it always reflects the state at the time of the call. NiceGUI's on_disconnect hook cleans up the subscription when the window closes. Any component that subscribes is notified — there is no mechanism by which a view can be forgotten.

users_page is three lines and holds no state of its own.

Scaling to Larger Domains

The current AppState is a concrete class, but the infrastructure around it is not. Moving to a Generic[T] AppStateWrapper and a SetAppStateCommand[T] means the same mutex, hash-check, validation, and notification behaviors apply to any state type. A component subscribed sole to AppStateWrapper[SessionState] is not re-rendered when AppStateWrapper[UserSettingsState] changes, because it holds a reference to a different wrapper instance. Cross-cutting pipeline behaviors are written once and apply to all state types. The only per-type work is defining the state class itself and registering an initial instance with the injector.

Multi-User and Multi-Tenant

Another dimension of scaling would be (multiple) concurrent users. Either you have an app with multiple users that isolates certain states and might share others (e.g., users from the same company might share data). The example project is intended to run in NiceGUI's "native" mode, which opens a webview as an application—the intended use is a single user running both frontend and backend locally. Still, I will briefly describe how to adjust our approach so multiple users could access the backend in parallel without running into bugs because of state leakage.

The current AppStateWrapper is a process-wide singleton — every connected session reads from and writes to the same instance. For purely shared state (a company-wide notification, a live feed) that is intentional. For session-private state like selected_user, each connection needs its own wrapper.

The cleanest extension point is the inject layer. Rather than binding a single instance, inject_application can register a factory backed by a session dictionary keyed on the client identifier:

_session_wrappers: dict[str, AppStateWrapper] = {}

def get_session_wrapper(client_id: str) -> AppStateWrapper:
    return _session_wrappers.setdefault(client_id, AppStateWrapper(AppState()))

Components call get_session_wrapper(ui.context.client.id) instead of inject.instance(AppStateWrapper), and the on_disconnect hook removes the entry when the session ends, releasing its subscribers and state.

The security boundary matters here. NiceGUI is a server-side rendering framework — Python objects, wrapper instances, and subscriber lists all live in server memory, pushed to the browser over a WebSocket. In client-side frameworks (React, Vue) each user's state lives in their own browser JavaScript heap and cross-user leakage via shared server objects is structurally impossible. With SSR, a mistakenly shared wrapper reference directly exposes one session's data to another, which makes explicit session scoping a correctness requirement rather than an optimisation.