Observable¶

The core observable class that provides reactive values.

A reactive value that automatically notifies dependents when it changes.

Observable is the core primitive of FynX's reactivity system. It wraps a value and provides transparent reactive behavior - when the value changes, all dependent computations and reactions are automatically notified and updated.

Key Features: - Transparent: Behaves like a regular value but with reactive capabilities - Dependency Tracking: Automatically tracks which reactive contexts depend on it - Change Notification: Notifies all observers when the value changes - Type Safety: Generic type parameter ensures type-safe operations - Lazy Evaluation: Computations only re-run when actually needed - Circular Dependency Detection: Prevents infinite loops at runtime

Observable implements various magic methods (__eq__, __str__, etc.) to behave like its underlying value in most contexts, making it easy to use in existing code without modification.

Attributes:

Name	Type	Description
`key`	`Optional[str]`	Unique identifier for debugging and serialization
`_value`	`Optional[T]`	The current wrapped value
`_observers`	`Set[Callable]`	Set of observer functions

Class Attributes

_current_context (Optional[ReactiveContext]): Current reactive execution context _context_stack (List[ReactiveContext]): Stack of nested reactive contexts

Parameters:

Name	Type	Description	Default
`key`	`Optional[str]`	A unique identifier for this observable (used for debugging). If None, will be set to "" and updated in set_name.	`None`
`initial_value`	`Optional[T]`	The initial value to store. Can be any type.	`None`

Raises:

Type	Description
`RuntimeError`	If a circular dependency is detected during value updates.

Example

from fynx.observable import Observable

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

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

# Subscribe to changes
def on_change():
    print(f"Counter changed to: {counter.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.

Accessing the value property automatically registers this observable as a dependency if called within a reactive context (computation or reaction).

Returns:

Type	Description
`Optional[T]`	The current value stored in this observable, or None if not set.

Note

This property is tracked by the reactive system. Use it instead of accessing _value directly to ensure proper dependency tracking.

Example

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)

Equality comparison with another value or observable.

Compares the wrapped values for equality. If comparing with another Observable, compares their wrapped values.

Parameters:

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

Returns:

Type	Description
`bool`	True if the values are equal, False otherwise.

Example

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

obs1 == obs2      # True (both wrap 5)
obs1 == regular_val  # True (observable equals regular value)
obs1 == 10        # False (5 != 10)

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])

# These will have different hashes despite same value
hash(obs1) != hash(obs2)  # 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 implements the descriptor protocol to enable automatic conversion of Observable instances to appropriate descriptors based on the owning class type.

For Store classes, the conversion is handled by StoreMeta metaclass. For other classes, converts to SubscriptableDescriptor for class-level observable behavior.

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. It modifies the class to use the appropriate descriptor for reactive behavior.

Example

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

# Gets converted to a descriptor automatically
instance = MyClass()
print(instance.obs)  # Uses descriptor

str ¶

__str__()

String representation of the wrapped value.

Returns the string representation of the current value, enabling observables to be used seamlessly in string contexts.

Returns:

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

Example

obs = Observable("name", "Alice")
print(f"Hello {obs}")  # Prints: "Hello Alice"
message = "User: " + obs  # Works like "User: " + obs.value

add_observer ¶

add_observer(observer)

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

Parameters:

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

remove_observer ¶

remove_observer(observer)

Remove an observer function.

Parameters:

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

set ¶

set(value)

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

This method updates the observable's value and triggers change notifications to all registered observers. The update only occurs if the new value is different from the current value (using != comparison).

Circular dependency detection is performed to prevent infinite loops where a computation tries to modify one of its own dependencies.

Parameters:

Name	Type	Description	Default
`value`	`Optional[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

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 using != operator, so custom objects should implement proper equality comparison if needed.

subscribe(func)

Subscribe a function to react to changes in this observable.

The subscribed function will be called whenever the observable's value changes.

Parameters:

Name	Type	Description	Default
`func`	`Callable`	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

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`	`Callable`	The function to unsubscribe from this observable	required

Observable¶

key ¶

value ¶

__bool__ ¶

__eq__ ¶

__hash__ ¶

__repr__ ¶

__set_name__ ¶

__str__ ¶

add_observer ¶

remove_observer ¶

set ¶

subscribe ¶

unsubscribe ¶

bool ¶

eq ¶

hash ¶

repr ¶

str ¶