Observable¶

The core observable class that provides reactive values.

A reactive value that notifies dependents automatically when it changes.

Wrap any Python value in an Observable, and functions that read it during reactive execution (via a ReactiveContext) will re-run when it changes. Reading .value inside a reactive context registers the observable as a dependency and adds the context's re-run function as an observer; calling .set() later notifies those observers with the fresh value.

Observable also implements __bool__ and __str__, so it can be used in boolean contexts (if observable:), string formatting (f"{observable}"), without reaching for .value explicitly. Equality is identity-based so observables remain safe in dependency sets and dicts; compare .value explicitly for value equality.

Changes are batched and notified in topological order - source observables first, then computed, then conditional - so a conditional observable never checks a condition value before it's been updated.

Circular dependencies raise a RuntimeError rather than looping forever: if a computation tries to modify one of its own dependencies (directly or indirectly), setting the value fails instead of recursing.

Attributes:

Name	Type	Description
`key`	`str`	Unique identifier for debugging and serialization
`_value`		The current wrapped value
`_observers`	`Set[Observer]`	Set of observer functions to notify on change

Class Attributes

_current_context: Current reactive execution context (None when not in reactive execution) _context_stack: Stack of nested reactive contexts for proper dependency tracking _pending_notifications: Set of observables waiting to notify observers _notification_scheduled: Whether notification processing is scheduled _currently_notifying: Set of observables currently notifying (prevents re-entrant notifications)

Parameters:

Name	Type	Description	Default
`key`	`Optional[str]`	A unique identifier for this observable (used for debugging and serialization). If None, will be set to "" and updated in set_name when used as a class attribute.	`None`
`initial_value`	`Optional[T]`	The initial value to store. Can be any type compatible with the generic type parameter.	`None`

Raises:

Type	Description
`RuntimeError`	If setting this value would create a circular dependency (e.g., a computed value trying to modify its own input).

Example

from fynx.observable import Observable

# Create an observable
counter = Observable("counter", 0)

# Direct access (transparent behavior)
print(counter.value)  # 0
print(counter.value == 0)  # True
print(str(counter))   # "0"

# Subscribe to changes
def on_change(new_value):
    print(f"Counter changed to: {new_value}")

counter.subscribe(on_change)
counter.set(5)  # Prints: "Counter changed to: 5"

Note

While you can create Observable instances directly, it's often more convenient to use the observable() descriptor in Store classes for better organization and automatic serialization support.

key ¶

key

Get the unique identifier for this observable.

value ¶

value

Get the current value of this observable.

Outside a reactive context this is a plain read. Inside one, reading it also registers this observable as a dependency of the current ReactiveContext, so a later .set() will re-run whatever depends on it.

Returns:

Type	Description
`T`	The current value stored in this observable.

Note

Always read through .value rather than _value directly - the dependency tracking depends on it.

Example

from fynx.observable import Observable
from fynx import reactive

obs = Observable("counter", 5)
print(obs.value)  # 5

# In a reactive context, this creates a dependency
@reactive(obs)
def print_value(val):
    print(f"Value: {val}")

bool ¶

__bool__()

Boolean conversion returns whether the value is truthy.

This allows observables to be used directly in boolean contexts (if statements, boolean operations) just like regular values.

Returns:

Type	Description
`bool`	True if the wrapped value is truthy, False otherwise.

Example

obs = Observable("flag", True)
if obs:  # Works like if obs.value
    print("Observable is truthy")

obs.set(0)  # False
if not obs:  # Works like if not obs.value
    print("Observable is falsy")

eq ¶

__eq__(other)

Identity comparison with another object.

Observables are mutable reactive graph nodes, so equality is based on node identity rather than wrapped value. This preserves Python's equality/hash contract when observables are used in dependency sets and dictionaries. Compare .value explicitly for value equality.

Parameters:

Name	Type	Description	Default
`other`	`object`	Value or Observable to compare with	required

Returns:

Type	Description
`bool`	True only when `other` is this same observable object.

Example

obs1 = Observable("a", 5)
obs2 = Observable("b", 5)
regular_val = 5

obs1 == obs2          # False (different graph nodes)
obs1.value == obs2.value  # True
obs1.value == regular_val # True

format ¶

__format__(format_spec)

Format the wrapped value with Python's normal format protocol.

get ¶

__get__(instance: None, owner: Type) -> 'ObservableValue[T]'

__get__(instance: object, owner: Type) -> 'ObservableValue[T]'

__get__(instance, owner=None)

Return a typed observable value wrapper when used as a class descriptor.

Standalone Observable instances keep their normal behavior. When an Observable is placed on a class, descriptor access returns ObservableValue[T], which gives static type checkers the same shape users interact with at runtime.

hash ¶

__hash__()

Hash based on object identity, not value.

Since values may be unhashable (like dicts, lists), observables hash based on their object identity rather than their value.

Returns:

Type	Description
`int`	Hash of the observable's object identity.

Note

This means observables with the same value will not be considered equal for hashing purposes, only identical objects.

Example

obs1 = Observable("a", [1, 2, 3])
obs2 = Observable("b", [1, 2, 3])

# Hashing follows graph-node identity, not wrapped value
hash(obs1) == id(obs1)  # True

# But identical objects hash the same
hash(obs1) == hash(obs1)  # True

repr ¶

__repr__()

Developer representation showing the observable's key and current value.

Returns:

Type	Description
`str`	A string representation useful for debugging and development.

Example

obs = Observable("counter", 42)
print(repr(obs))  # Observable('counter', 42)

__set_name__ ¶

__set_name__(owner, name)

Called when this Observable is assigned to a class attribute.

This method records the descriptor name and defining owner. Observable implements the descriptor protocol directly: class access returns an ObservableValue[T] wrapper, and StoreMeta records the owner-specific backing Observable for class-level assignment.

Parameters:

Name	Type	Description	Default
`owner`	`Type`	The class that owns this attribute	required
`name`	`str`	The name of the attribute being assigned	required

Note

This method is called automatically by Python when an Observable instance is assigned to a class attribute.

Example

class MyClass:
    obs = Observable("counter", 0)  # __set_name__ called here

instance = MyClass()
print(instance.obs)  # Uses descriptor

str ¶

__str__()

String representation of the wrapped value.

Lets an observable drop straight into string contexts (f-strings, str()) without unwrapping it first.

Returns:

Type	Description
`str`	String representation of the wrapped value.

Example

obs = Observable("name", "Alice")
print(f"Hello {obs}")  # Prints: "Hello Alice"
# Note: String concatenation with + requires explicit .value access
message = "User: " + str(obs)  # Works with str() conversion

add_fast_observer ¶

add_fast_observer(observer)

Add an internal source-only observer that receives the new value.

add_observer ¶

add_observer(observer)

Add an observer function that will be called when this observable changes.

Parameters:

Name	Type	Description	Default
`observer`	`Observer`	A callable that takes no arguments	required

remove_fast_observer ¶

remove_fast_observer(observer)

Remove an internal source-only observer.

remove_observer ¶

remove_observer(observer)

Remove an observer function.

Parameters:

Name	Type	Description	Default
`observer`	`Observer`	The observer function to remove	required

set ¶

set(value)

Set the value and notify all observers if the value changed.

The update only happens if the new value differs from the current one (via !=); if it's unchanged, observers aren't notified and no recomputation happens.

Before updating, this checks whether the current reactive context already depends on this observable - if so, the set would create a cycle, and a RuntimeError is raised instead.

When the value does change, notifications are queued and processed in topological order (source, then computed, then conditional), so a conditional observable never sees a stale condition value.

Parameters:

Name	Type	Description	Default
`value`	`T`	The new value to set. Can be any type compatible with the observable's generic type parameter.	required

Raises:

Type	Description
`RuntimeError`	If setting this value would create a circular dependency (e.g., a computed value trying to modify its own input).

Example

from fynx.observable import Observable

obs = Observable("counter", 0)
obs.set(5)  # Triggers observers if value changed

# No change, no notification
obs.set(5)  # Same value, observers not called

Note

Equality is checked defensively. If a value's equality operator raises or returns a non-boolean comparison object that cannot be coerced safely, FynX treats the assignment as changed and notifies.

subscribe(func)

Subscribe a function to react to changes in this observable.

The function is called with the new value whenever .set() changes it - not immediately on subscription. Internally this wraps func in a ReactiveContext, so if it also reads other observables, it'll re-run when those change too.

Parameters:

Name	Type	Description	Default
`func`	`Subscriber[T]`	A callable that accepts one argument (the new value). The function will be called whenever the observable's value changes.	required

Returns:

Type	Description
`'Observable[T]'`	This observable instance for method chaining.

Example

from fynx.observable import Observable

def on_change(new_value):
    print(f"Observable changed to: {new_value}")

obs = Observable("counter", 0)
obs.subscribe(on_change)

obs.set(5)  # Prints: "Observable changed to: 5"

Note

The function is called only when the observable's value changes. It is not called immediately upon subscription.

unsubscribe ¶

unsubscribe(func)

Unsubscribe a function from this observable.

Parameters:

Name	Type	Description	Default
`func`	`Subscriber[T]`	The function to unsubscribe from this observable	required

Observable¶

key ¶

value ¶

__bool__ ¶

__eq__ ¶

__format__ ¶

__get__ ¶

__hash__ ¶

__repr__ ¶

__set_name__ ¶

__str__ ¶

add_fast_observer ¶

add_observer ¶

remove_fast_observer ¶

remove_observer ¶

set ¶

subscribe ¶

unsubscribe ¶

bool ¶

eq ¶

format ¶

get ¶

hash ¶

repr ¶

str ¶