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, which implements this tutorial's minimum specification. The current 3.25.0 Python source line is not published yet; pin vmx==3.23.0 when reproducing 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)
    # Source 3.23.2: foreground → AsyncIOThreadSafeScheduler(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:

import asyncio

from reactivex.scheduler import ThreadPoolScheduler
from reactivex.scheduler.eventloop import AsyncIOThreadSafeScheduler
from vmx.services import RxDispatcher

async def main() -> None:
    loop = asyncio.get_running_loop()
    # Verified workaround for published vmx 3.23.0 with RxPY 4.0.4.
    dispatcher = RxDispatcher(
        foreground=AsyncIOThreadSafeScheduler(loop),
        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 is AsyncIOThreadSafeScheduler(loop), which safely posts worker completions back to the given event loop. The host still owns the loop and the independent background pool; see Python asyncio dispatcher ownership for teardown guidance.

3.3.7.1. Start And Cancel An Async Resource From Synchronous Code

load_command.execute() may be called when no asyncio loop is running on the caller. VMx starts the resource operation on its shared daemon loop. A later cancel() invalidates the resource state synchronously and posts native task/future work to that operation loop, so command settlement may follow the state change:

This recipe requires the unreleased Python 3.25.0 source installed from a checkout. The current PyPI 3.23.0 package does not contain this operation-loop repair.

import asyncio
import threading

from vmx import (
    NULL_DISPATCHER,
    AsyncResourceStatus,
    AsyncResourceVM,
    MessageHub,
)

started = threading.Event()
cancelled = threading.Event()


async def load_report() -> str:
    started.set()
    try:
        await asyncio.Future()
    except asyncio.CancelledError:
        cancelled.set()
        raise


hub = MessageHub()
report = AsyncResourceVM(
    name="report",
    loader=load_report,
    hub=hub,
    dispatcher=NULL_DISPATCHER,
)

report.load_command.execute()
assert started.wait(3)
report.cancel()
assert report.state.status is AsyncResourceStatus.IDLE
assert cancelled.wait(3)

report.dispose()
hub.dispose()

The corresponding langs/python/tests/unit/state/test_async_resource_threading.py also records the loop used for task cancellation, future completion, and late callback registration. State and cleanup notifications are not automatically UI-dispatched. For caller-owned loops, stop admissions, cancel, and drain while the loop can still run before closing it; closed-loop pending work cannot be completed by VMx. Do not close VMx's shared daemon loop. See the central asyncio ownership guidance.

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