3.3. Getting Started with VMx — Python

This tutorial walks you through building viewmodels with the VMx Python library. You will build a ComponentVMOf[UserModel], a RelayCommand with a reactive trigger, and a CompositeVM[TabVM] with tab selection — all in a Python REPL, script, or test.

For the contracts behind each type, see the component family, command families, and composite family.

---

3.3.1. Install

PyPI provides VMx 3.23.0, matching this tutorial's Python source line and minimum specification. Pin vmx==3.23.0 when reproducing this released behavior.

# Using uv (recommended)
uv add vmx

# Using pip
pip install vmx

For local development from a checked-out clone:

uv add --editable path/to/VMx/langs/python
# or
pip install -e path/to/VMx/langs/python

---

3.3.2. Wire up MessageHub and RxDispatcher

Every viewmodel needs two services: a hub that carries messages between viewmodels and a dispatcher that knows about your event loop or UI thread.

3.3.2.1. Option A — immediate (console / synchronous tests)

from vmx.services import MessageHub, RxDispatcher

hub = MessageHub()
dispatcher = RxDispatcher.immediate()
# Both foreground and background schedulers are ImmediateScheduler — safe for
# console scripts and pytest suites where there is no event loop.

3.3.2.2. Option B — asyncio-based UI (Textual, etc.)

import asyncio
from vmx.services import MessageHub, RxDispatcher

async def main() -> None:
    loop = asyncio.get_running_loop()
    hub = MessageHub()
    dispatcher = RxDispatcher.asyncio(loop)
    # foreground → AsyncIOScheduler(loop)
    # background → ThreadPoolScheduler
    ...

asyncio.run(main())

asyncio.get_running_loop() is preferred over asyncio.get_event_loop(), which has been a DeprecationWarning since Python 3.10 when no loop is running.

You can also inject the two schedulers directly if you need a custom pairing:

from reactivex.scheduler import ImmediateScheduler, ThreadPoolScheduler
from vmx.services import RxDispatcher

dispatcher = RxDispatcher(
    foreground=ImmediateScheduler(),
    background=ThreadPoolScheduler(),
)

---

3.3.3. Build a ComponentVMOf[UserModel]

ComponentVMOf[M] is the primary leaf viewmodel. It holds a typed model, fires PropertyChangedMessage on the hub when the model changes, and participates in the lifecycle state machine (DESTRUCTED → CONSTRUCTING → CONSTRUCTED → DESTRUCTING → DESTRUCTED).

from dataclasses import dataclass

from vmx.components import ComponentVMOf
from vmx.messages import PropertyChangedMessage
from vmx.services import MessageHub, RxDispatcher

# A simple domain model — use your own real types here.
@dataclass(frozen=True)
class UserModel:
    name: str
    email: str

hub = MessageHub()
dispatcher = RxDispatcher.immediate()

# Build the viewmodel — every builder setter returns a NEW builder (immutable).
user_vm: ComponentVMOf[UserModel] = (
    ComponentVMOf.builder()
    .name("user-card")
    .model(UserModel("Alice", "alice@example.com"))
    .services(hub, dispatcher)
    # Derive a display hint from the model.
    .modeled_hinter(lambda m: m.name)
    # Optional: callback when model is set to a new value.
    .on_model_changed(lambda m: print(f"Model updated → {m.name}"))
    .on_construct(lambda: print("user-card constructed"))
    .on_destruct(lambda: print("user-card destructed"))
    .build()
)

# Subscribe to hub messages BEFORE constructing so you don't miss any.
# reactivex.operators has no OfType; filter by isinstance directly in the
# subscriber.
hub.messages.subscribe(
    lambda msg: (
        isinstance(msg, PropertyChangedMessage)
        and msg.sender is user_vm
        and print(f"Property '{msg.property_name}' changed on {msg.sender_name}")
    )
)

# Alternatively, subscribe via the VM's own property_changed observable,
# which emits property name strings (snake_case).
user_vm.property_changed.subscribe(
    lambda prop: print(f"  [property_changed] {prop}")
)

# construct() transitions DESTRUCTED → CONSTRUCTING → CONSTRUCTED.
# This fires the VM's on_construct callback. Model assignment is already valid
# before construction and remains valid until the terminal DISPOSED state.
user_vm.construct()
# stdout: "user-card constructed"

# Update the model — triggers on_model_changed and publishes
# PropertyChangedMessage for "model" (and "modeled_hint" if it changed).
user_vm.model = UserModel("Alice Smith", "asmith@example.com")
# stdout: "Property 'model' changed on user-card"
# stdout: "Model updated → Alice Smith"

print(user_vm.modeled_hint)  # "Alice Smith"  (modeled_hinter result)
print(user_vm.status)        # ConstructionStatus.CONSTRUCTED

See the component family for the full ComponentVMOfProto[M] contract and Services, Messages & Dispatching for the PropertyChangedMessage schema.

---

3.3.4. Build a RelayCommand

RelayCommand wraps an optional callable task, an optional predicate that gates execution, and a set of Observable triggers that signal can_execute may have changed.

from reactivex.subject import Subject

from vmx.commands import RelayCommand

# A Subject you fire whenever the predicate outcome may have changed.
can_save_trigger: Subject[object] = Subject()

is_dirty = False

def save_task() -> None:
    global is_dirty
    print("Saving…")
    is_dirty = False
    can_save_trigger.on_next(None)  # re-evaluate can_execute

save_command = (
    RelayCommand.builder()
    .task(save_task)
    .predicate(lambda: is_dirty)
    .triggers(can_save_trigger)
    .build()
)

# can_execute is False until is_dirty is True.
print(save_command.can_execute())   # False

is_dirty = True
can_save_trigger.on_next(None)      # fires can_execute_changed

# Subscribe to re-evaluation notifications.
save_command.can_execute_changed.subscribe(
    lambda _: print(f"  can_execute is now {save_command.can_execute()}")
)

print(save_command.can_execute())   # True
save_command.execute()              # prints "Saving…"
print(save_command.can_execute())   # False again

# Dispose to unsubscribe all trigger subscriptions.
save_command.dispose()

See command families for the full command contract, including the "predicate-false gates execute" rule (CMD-003).

---

3.3.5. Build a CompositeVM[TabVM]

CompositeVM[VM] owns an ordered child collection and a current selection. Children are provided by a factory callable that runs lazily on the first construct() call.

from dataclasses import dataclass

from vmx.components import ComponentVMOf
from vmx.composites import CompositeVM
from vmx.messages import PropertyChangedMessage
from vmx.services import MessageHub, RxDispatcher

@dataclass(frozen=True)
class TabModel:
    title: str

hub = MessageHub()
dispatcher = RxDispatcher.immediate()

# Build two tab children — they share the same hub and dispatcher.
tab1: ComponentVMOf[TabModel] = (
    ComponentVMOf.builder()
    .name("home-tab")
    .model(TabModel("Home"))
    .services(hub, dispatcher)
    .build()
)

tab2: ComponentVMOf[TabModel] = (
    ComponentVMOf.builder()
    .name("settings-tab")
    .model(TabModel("Settings"))
    .services(hub, dispatcher)
    .build()
)

# Build the composite. The children factory is evaluated on construct().
tabs: CompositeVM[ComponentVMOf[TabModel]] = (
    CompositeVM.builder()
    .name("tab-bar")
    .services(hub, dispatcher)
    .children(lambda: [tab1, tab2])
    .on_construct(lambda: print("tab-bar ready"))
    .build()
)

# Watch for current-selection changes via the hub.
hub.messages.subscribe(
    lambda msg: (
        isinstance(msg, PropertyChangedMessage)
        and msg.sender is tabs
        and msg.property_name == "current"
        and print(
            f"Selected tab: {tabs.current.model.title if tabs.current else '(none)'}"
        )
    )
)

# construct() cascades: the composite constructs itself then each child.
tabs.construct()
# stdout: "tab-bar ready"

# Select a tab — publishes PropertyChangedMessage for "current" and
# sets child.is_current.
tabs.current = tab2   # stdout: "Selected tab: Settings"
tabs.current = tab1   # stdout: "Selected tab: Home"

print([child.name for child in tabs])  # ['home-tab', 'settings-tab']
print(tab2.is_current)                 # False

See the composite family for the full CompositeVMProto[VM] contract, including the MutableSequence semantics and CollectionChangedEvent.

---

3.3.6. Lifecycle and cleanup

Every VM follows a five-state lifecycle: DESTRUCTED → CONSTRUCTING → CONSTRUCTED → DESTRUCTING → DESTRUCTED, plus the terminal DISPOSED.

from vmx.lifecycle.status import ConstructionStatus

print(user_vm.status)    # ConstructionStatus.CONSTRUCTED  (after construct())

# reconstruct() is destruct() + construct() in a single call. It is only valid
# from CONSTRUCTED (can_reconstruct() is True iff status == CONSTRUCTED); it
# round-trips through DESTRUCTED and back to CONSTRUCTED.
user_vm.reconstruct()
print(user_vm.status)    # ConstructionStatus.CONSTRUCTED

# destruct() transitions back to DESTRUCTED and runs on_destruct.
user_vm.destruct()
print(user_vm.status)    # ConstructionStatus.DESTRUCTED

# dispose() is terminal and idempotent. Calling construct() or destruct() on a
# disposed VM raises StatusTransitionError.
user_vm.dispose()
print(user_vm.status)    # ConstructionStatus.DISPOSED

# CompositeVM.dispose() disposes children, then itself.
tabs.dispose()

# MessageHub.dispose() completes the underlying Rx Subject.
hub.dispose()

See Lifecycle & Messaging for the full transition table and the lifecycle contract (LIFE-001..015), including StatusTransitionError rules and admitted-hook/disposal coordination.

---

3.3.7. Threading

RxDispatcher pairs two Rx schedulers:

Scheduler	Typical mapping
dispatcher.foreground	UI thread / asyncio event loop
dispatcher.background	Thread-pool (blocking I/O, CPU-bound work)

All hub observations delivered on foreground are safe to bind to UI controls. Use observe_on from reactivex.operators to marshal:

import reactivex.operators as ops

hub.messages.pipe(
    ops.filter(lambda msg: isinstance(msg, PropertyChangedMessage)),
    ops.observe_on(dispatcher.foreground),   # marshal to UI scheduler
).subscribe(lambda msg: update_label(msg))   # safe to touch UI here

For background work before constructing a VM:

import reactivex as rx
import reactivex.operators as ops

def apply_remote_data(data: UserModel) -> None:
    user_vm.model = data
    user_vm.construct()

rx.from_callable(lambda: load_from_database(), scheduler=dispatcher.background).pipe(
    ops.observe_on(dispatcher.foreground),
).subscribe(apply_remote_data)

When using RxDispatcher.asyncio(loop), the foreground scheduler posts work back to the given asyncio event loop, keeping VM mutations on the loop thread.

See Services, Messages & Dispatching for the THR-001..THR-004 conformance rules.

---

3.3.8. Where to go next

Resource	Documentation page
Specification status	Specification & Conformance
Lifecycle contract	Lifecycle & Messaging
Messages & threading	Services, Messages & Dispatching
Commands	Command Families
Component contract	Component Family
Composite contract	Composite Family
Builders & tree	Builders, Collections & Tree Utilities
Architecture	Architecture Map
Python status	Python Flavor
Examples	Smaller Examples