robject

Reactive objects for Python. Declare a class the way you would a dataclass, and its attributes become observable. Derived values recompute only when what they read actually changed, side effects re-run automatically, and a burst of writes can be grouped into a single notification.

It's the signals / computed / effect model from modern UI frameworks, packaged as plain Python classes. Pure standard library, no dependencies.

from robject import RObject, computed, effect, batch

class Cart(RObject):
    price: float = 0.0
    qty: int = 1

    @computed
    def total(self) -> float:
        return self.price * self.qty

cart = Cart(price=2.5)
effect(lambda: print("total is", cart.total))   # prints: total is 2.5

with batch():
    cart.price = 3.0
    cart.qty = 4
# prints once: total is 12.0

Features

• Reactive fields from ordinary class annotations, with defaults, required fields, field(default_factory=...) and inheritance.
• @computed properties: lazy, cached and read-only. Dependencies are tracked automatically, including dynamic ones (branches) and ones on other objects. Exceptions are cached too, and cycles are detected.
• Effects re-run whenever anything they read changes. They are glitch-free: an effect never sees a half-updated state.
• Subscriptions deliver Change(obj, name, old, new) events for fields and computed properties. A computed property notifies only when its value really changes.
• Batching coalesces writes. Subscribers get one event per attribute (none if the value ends up back where it started), and effects run once.
• Safety rails: an error in one callback doesn't stop the others (the first one is re-raised afterwards), and runaway feedback loops raise instead of hanging.
• Typed (py.typed), standard library only, Python 3.10+.

Install

From a checkout (the package is not published to PyPI):

python3 -m venv .venv && . .venv/bin/activate
pip install .

Quickstart

from robject import RObject, Change, batch, computed, effect, field, subscribe

class Todo(RObject):
    title: str                                 # required
    done: bool = False
    tags: list = field(default_factory=list)   # mutable defaults need a factory

class TodoList(RObject):
    items: tuple = ()

    @computed
    def remaining(self) -> int:
        return sum(not t.done for t in self.items)

todos = TodoList(items=(Todo(title="write docs"), Todo(title="ship")))

# Subscribe to specific attributes (or all of them, by passing no names).
unsubscribe = subscribe(todos, lambda ch: print(f"{ch.name}: {ch.old} -> {ch.new}"), "remaining")

todos.items[0].done = True          # prints: remaining: 2 -> 1

# Effects track whatever they read, across objects.
watcher = effect(lambda: print("left:", todos.remaining))   # prints: left: 1

with batch():                       # delivery waits until the block ends
    todos.items[1].done = True      # remaining: 1 -> 0 ...
    todos.items = todos.items + (Todo(title="celebrate"),)   # ... -> 1
# The effect re-runs once and prints: left: 1
# `remaining` ended where it started, so the subscriber prints nothing.

watcher.dispose()
unsubscribe()

Fields are tracked by assignment. Mutating a list in place (todo.tags.append(...)) is invisible, so assign a new value instead (todo.tags = [*todo.tags, "x"]), or use immutable containers such as tuples.

API overview

Everything below is importable from robject.

Semantics worth knowing

• A write that is == to the current value is a no-op.
• Outside a batch, delivery is synchronous: by the time an assignment returns, subscribers and effects have run.
• Subscribers and effects may write to reactive values. Their work is processed in further rounds of the same flush. If 100 rounds pass without settling, the flush raises RuntimeError("reactive updates did not settle ...").
• An effect's own reads are tracked; reads made by callbacks it triggers are not.
• The reactive graph is process-global and not thread-safe. Use it from one thread, such as a UI or event-loop thread.

Development

python3 -m unittest discover -s tests -v

The tests use only unittest, so python3 -m pytest works too if you have pytest.

License

MIT