API Reference

Everything documented on this page is available from the top-level observ package:

from observ import (
    reactive, readonly, shallow_reactive, shallow_readonly, ref, to_raw, trigger_ref,
    computed, watch, watch_effect, Watcher,
    init, loop_factory, scheduler,
)

Reactive state

reactive

reactive(target: T) -> T

Returns a reactive proxy for the given target, which behaves like the target itself but tracks reads and writes for dependency tracking. Supports (arbitrarily nested) dicts, lists, sets and tuples; plain values are returned as-is. Calling reactive multiple times on the same target returns the same proxy. See Reactivity.

readonly

readonly(target: T) -> T

Same as reactive, but the returned proxy can only be read from: reads are tracked, and any write raises ReadonlyError. See Readonly and Shallow Proxies.

shallow_reactive

shallow_reactive(target: T) -> T

Same as reactive, but only the first level of the target is made reactive: nested values are returned raw. See Readonly and Shallow Proxies.

shallow_readonly

shallow_readonly(target: T) -> T

Combination of shallow_reactive and readonly.

observ.proxy.ref

ref(target)

Returns a reactive dict with a single 'value' key, set to the given target. Useful for making a single (plain) value reactive.

observ.proxy.trigger_ref

trigger_ref(target)

Force-notify the watchers that depend on the given proxy, as if its first level was written to. This is typically used together with a shallow proxy (shallow_reactive, shallow_readonly), after making deep mutations to a value nested inside it — those mutations bypass the proxy, so observ cannot see them by itself.

Note that the parameter is annotated Proxy[T] | T (like to_raw) rather than just Proxy[T]: proxies are typed as their target's own type, so requiring Proxy would reject every correctly-typed call site. A proxy is still required at runtime.

observ.proxy.to_raw

to_raw(target)

Returns a raw object from which any trace of proxy has been replaced with its wrapped target value.

Watching state

observ.watcher.watch

watch(fn, callback=None, sync=False, deep=None, immediate=False)

Watch the given function (or proxy) and call the optional callback when its value changes. Returns a Watcher object: keep a reference to it to keep the watcher alive, and use it to stop, pause and resume watching.

Function to watch. Can also be a proxy (or list of proxies),

which implies deep watching.

callback: Method to call when the watched value has changed. May accept zero, one (new value) or two (new and old value) arguments. When no callback is given, fn is re-evaluated when its dependencies change (see also watch_effect). sync: Run the callback immediately on change instead of queueing it on the scheduler. deep: Also watch for changes nested inside the watched value. Defaults to False when fn is callable, True otherwise. immediate: Call the callback right away with the initial value.

observ.watcher.watch_effect

watch_effect(fn, sync=False, deep=True)

Run the given function immediately to collect its dependencies and re-run it whenever they change. Equivalent to calling watch without a callback.

observ.watcher.computed

computed(_fn: Callable[[], T]) -> Computed[T]

computed(*, deep: bool = True) -> Callable[[Callable[[], T]], Computed[T]]

computed(_fn=None, *, deep=True)

Create derived state from a function: the result is cached and only recomputed (lazily) when any of the reactive state it depends on has changed. Can be used as a (parameterized) decorator.

Make sure fn doesn't need any arguments to run and that no reactive state is changed within the function.

observ.watcher.Watcher

active property

active

Returns whether this watcher is still active. To manually deactivate simply call the watcher object.

paused property

paused

Returns whether this watcher is currently paused. Use pause() and resume() to control this state.

stop

stop()

Stop the watcher and clean up any used resource. Equivalent to calling the watcher object directly.

pause

pause()

Temporarily pause the watcher: while paused, changes to dependencies will not trigger a re-evaluation or callback. Resume the watcher with resume().

resume

resume()

Resume the watcher after it was paused. If any of its dependencies changed while the watcher was paused, it will trigger once upon resume.

Scheduling

observ.init.init

init(mode='asyncio', loop=None)

Integrate the scheduler with an event loop, so that watcher callbacks are batched and run on the loop. Supported modes are 'asyncio' (the default), 'qt' and 'rendercanvas'. A specific loop object can be given for the 'asyncio' and 'rendercanvas' modes.

observ.init.loop_factory

loop_factory()

Creates a new asyncio event loop with the eager task factory enabled, which reduces the latency of the tasks that observ schedules for async functions and callbacks.

scheduler

The global Scheduler instance on which watchers are queued. Use the register_* methods (or the init() shorthand) to integrate it with an event loop, or call flush() manually. See Scheduling.

observ.scheduler.Scheduler

register_asyncio

register_asyncio(loop=None)

Utility function for integration with asyncio.

If no loop object is given, get_event_loop() is used on each flush to determine the current loop.

register_qt

register_qt()

Legacy utility function for integration with Qt event loop. Note that using the register_asyncio method is preferred over this, together with running QtAsyncio.run(...) instead of app.exec(). This is supported from Pyside 6.7. Note that the QtAsyncio submodule is not included in the pyside6_essentials package.

register_rendercanvas

register_rendercanvas(loop)

Utility function for integration with rendercanvas loop objects

register_request_flush

register_request_flush(callback)

Register callback for registering a call to flush

flush

flush()

Flush the queue to evaluate all queued watchers. You can call this manually, or register a callback to request to perform the flush.