Skip to content

Observable

observant.observable.Observable

Bases: Generic[T], IObservable[T]

A generic observable value that notifies listeners when its value changes.

Observable is the core building block of Observant's reactive system. It wraps a single value and provides methods to get, set, and observe changes to that value.

Attributes:

Name Type Description
_value T

The current value of the observable.
_callbacks list[Callable[[T], None]]

List of callback functions to be called when the value changes.
_on_change_enabled bool

Whether callbacks are enabled.

Examples:

# Create an observable integer
counter = Observable[int](0)

# Register a callback to be notified when the value changes
counter.on_change(lambda value: print(f"Counter changed to {value}"))

# Update the value
counter.set(1)  # Prints: "Counter changed to 1"

# Get the current value
current_value = counter.get()  # Returns: 1
Source code in observant\observable.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class Observable(Generic[T], IObservable[T]):
    """
    A generic observable value that notifies listeners when its value changes.

    Observable is the core building block of Observant's reactive system. It wraps
    a single value and provides methods to get, set, and observe changes to that value.

    Attributes:
        _value: The current value of the observable.
        _callbacks: List of callback functions to be called when the value changes.
        _on_change_enabled: Whether callbacks are enabled.

    Examples:
        ```python
        # Create an observable integer
        counter = Observable[int](0)

        # Register a callback to be notified when the value changes
        counter.on_change(lambda value: print(f"Counter changed to {value}"))

        # Update the value
        counter.set(1)  # Prints: "Counter changed to 1"

        # Get the current value
        current_value = counter.get()  # Returns: 1
        ```
    """

    _value: T
    _callbacks: list[Callable[[T], None]]
    _on_change_enabled: bool = True

    def __init__(self, value: T, *, on_change: Callable[[T], None] | None = None, on_change_enabled: bool = True) -> None:
        """
        Initialize the Observable with a value.

        Args:
            value: The initial value of the observable.
            on_change: Optional callback function to register immediately.
            on_change_enabled: Whether callbacks should be enabled initially.
        """
        print(f"DEBUG: Observable.__init__ called with value {value}")
        self._value = value
        self._callbacks = []
        self._on_change_enabled = on_change_enabled
        print("DEBUG: Observable.__init__ - Initialized with empty callbacks list")

    @override
    def get(self) -> T:
        """
        Get the current value of the observable.

        Returns:
            The current value stored in this observable.

        Examples:
            ```python
            counter = Observable[int](0)
            value = counter.get()  # Returns: 0
            ```
        """
        return self._value

    @override
    def set(self, value: T, notify: bool = True) -> None:
        """
        Set a new value for the observable and notify all registered callbacks.

        This method updates the internal value and, if notify is True and callbacks
        are enabled, calls all registered callbacks with the new value.

        Args:
            value: The new value to set.
            notify: Whether to notify the callbacks after setting the value.

        Examples:
            ```python
            counter = Observable[int](0)
            counter.on_change(lambda value: print(f"Counter changed to {value}"))

            # Update with notification
            counter.set(1)  # Prints: "Counter changed to 1"

            # Update without notification
            counter.set(2, notify=False)  # No output
            ```
        """
        print(f"DEBUG: Observable.set called with value {value}")
        self._value = value

        if not notify or not self._on_change_enabled:
            print("DEBUG: Observable.set - on_change is disabled, skipping callbacks")
            return

        print(f"DEBUG: Observable.set - Notifying {len(self._callbacks)} callbacks")
        for i, callback in enumerate(self._callbacks):
            print(f"DEBUG: Observable.set - Calling callback {i}")
            callback(value)
            print(f"DEBUG: Observable.set - Callback {i} completed")
        print("DEBUG: Observable.set - Completed")

    @override
    def on_change(self, callback: Callable[[T], None]) -> None:
        """
        Register a callback function to be called when the value changes.

        The callback will be called with the new value whenever set() is called
        with notify=True and callbacks are enabled. Callbacks are called in the
        order they were registered.

        If the same callback function is registered multiple times, it will only
        be added once.

        Args:
            callback: A function that takes the new value as its argument.

        Examples:
            ```python
            counter = Observable[int](0)

            # Register a callback
            counter.on_change(lambda value: print(f"Counter changed to {value}"))

            # Register another callback
            counter.on_change(lambda value: print(f"Counter is now {value}"))

            # Update the value
            counter.set(1)
            # Prints:
            # "Counter changed to 1"
            # "Counter is now 1"
            ```
        """
        print(f"DEBUG: Observable.on_change called, current callbacks: {len(self._callbacks)}")
        # Check if this callback is already registered to avoid duplicates
        for existing_cb in self._callbacks:
            if existing_cb == callback:
                print("DEBUG: Observable.on_change - Callback already registered, skipping")
                return

        self._callbacks.append(callback)
        print(f"DEBUG: Observable.on_change - Added callback, now have {len(self._callbacks)} callbacks")

    @override
    def enable(self) -> None:
        """
        Enable the observable to notify changes.

        After calling this method, subsequent calls to set() with notify=True
        will trigger callbacks.

        Examples:
            ```python
            counter = Observable[int](0)
            counter.on_change(lambda value: print(f"Counter changed to {value}"))

            # Disable notifications
            counter.disable()
            counter.set(1)  # No output

            # Enable notifications
            counter.enable()
            counter.set(2)  # Prints: "Counter changed to 2"
            ```
        """
        print("DEBUG: Observable.enable called")
        self._on_change_enabled = True

    @override
    def disable(self) -> None:
        """
        Disable the observable from notifying changes.

        After calling this method, subsequent calls to set() will not trigger
        callbacks, even if notify=True.

        Examples:
            ```python
            counter = Observable[int](0)
            counter.on_change(lambda value: print(f"Counter changed to {value}"))

            # Disable notifications
            counter.disable()
            counter.set(1)  # No output
            ```
        """
        print("DEBUG: Observable.disable called")
        self._on_change_enabled = False

    def __bool__(self) -> bool:
        """
        Convert the observable to a boolean.

        This allows using the observable directly in boolean contexts.

        Returns:
            The boolean value of the current value.

        Examples:
            ```python
            counter = Observable[int](0)
            if not counter:
                print("Counter is zero")  # This will print

            counter.set(1)
            if counter:
                print("Counter is non-zero")  # This will print
            ```
        """
        return bool(self.get())

    @override
    def __str__(self) -> str:
        """
        Convert the observable to a string.

        This allows using the observable directly in string contexts.

        Returns:
            The string representation of the current value.

        Examples:
            ```python
            counter = Observable[int](42)
            print(f"The counter is {counter}")  # Prints: "The counter is 42"
            ```
        """
        return str(self.get())

    @override
    def __repr__(self) -> str:
        """
        Get the representation of the observable.

        Returns:
            A string representation of the observable, including its class name
            and current value.

        Examples:
            ```python
            counter = Observable[int](42)
            repr(counter)  # Returns: "Observable(42)"
            ```
        """
        return f"{self.__class__.__name__}({self.get()!r})"

__bool__()

Convert the observable to a boolean.

This allows using the observable directly in boolean contexts.

Returns:

Type Description
bool

The boolean value of the current value.

Examples:

counter = Observable[int](0)
if not counter:
    print("Counter is zero")  # This will print

counter.set(1)
if counter:
    print("Counter is non-zero")  # This will print
Source code in observant\observable.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def __bool__(self) -> bool:
    """
    Convert the observable to a boolean.

    This allows using the observable directly in boolean contexts.

    Returns:
        The boolean value of the current value.

    Examples:
        ```python
        counter = Observable[int](0)
        if not counter:
            print("Counter is zero")  # This will print

        counter.set(1)
        if counter:
            print("Counter is non-zero")  # This will print
        ```
    """
    return bool(self.get())

__init__(value, *, on_change=None, on_change_enabled=True)

Initialize the Observable with a value.

Parameters:

Name Type Description Default
value T

The initial value of the observable.

 required
on_change Callable[[T], None] | None

Optional callback function to register immediately.

 None
on_change_enabled bool

Whether callbacks should be enabled initially.

 True
Source code in observant\observable.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(self, value: T, *, on_change: Callable[[T], None] | None = None, on_change_enabled: bool = True) -> None:
    """
    Initialize the Observable with a value.

    Args:
        value: The initial value of the observable.
        on_change: Optional callback function to register immediately.
        on_change_enabled: Whether callbacks should be enabled initially.
    """
    print(f"DEBUG: Observable.__init__ called with value {value}")
    self._value = value
    self._callbacks = []
    self._on_change_enabled = on_change_enabled
    print("DEBUG: Observable.__init__ - Initialized with empty callbacks list")

__repr__()

Get the representation of the observable.

Returns:

Type Description
str

A string representation of the observable, including its class name
str

and current value.

Examples:

counter = Observable[int](42)
repr(counter)  # Returns: "Observable(42)"
Source code in observant\observable.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
@override
def __repr__(self) -> str:
    """
    Get the representation of the observable.

    Returns:
        A string representation of the observable, including its class name
        and current value.

    Examples:
        ```python
        counter = Observable[int](42)
        repr(counter)  # Returns: "Observable(42)"
        ```
    """
    return f"{self.__class__.__name__}({self.get()!r})"

__str__()

Convert the observable to a string.

This allows using the observable directly in string contexts.

Returns:

Type Description
str

The string representation of the current value.

Examples:

counter = Observable[int](42)
print(f"The counter is {counter}")  # Prints: "The counter is 42"
Source code in observant\observable.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
@override
def __str__(self) -> str:
    """
    Convert the observable to a string.

    This allows using the observable directly in string contexts.

    Returns:
        The string representation of the current value.

    Examples:
        ```python
        counter = Observable[int](42)
        print(f"The counter is {counter}")  # Prints: "The counter is 42"
        ```
    """
    return str(self.get())

disable()

Disable the observable from notifying changes.

After calling this method, subsequent calls to set() will not trigger callbacks, even if notify=True.

Examples:

counter = Observable[int](0)
counter.on_change(lambda value: print(f"Counter changed to {value}"))

# Disable notifications
counter.disable()
counter.set(1)  # No output
Source code in observant\observable.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@override
def disable(self) -> None:
    """
    Disable the observable from notifying changes.

    After calling this method, subsequent calls to set() will not trigger
    callbacks, even if notify=True.

    Examples:
        ```python
        counter = Observable[int](0)
        counter.on_change(lambda value: print(f"Counter changed to {value}"))

        # Disable notifications
        counter.disable()
        counter.set(1)  # No output
        ```
    """
    print("DEBUG: Observable.disable called")
    self._on_change_enabled = False

enable()

Enable the observable to notify changes.

After calling this method, subsequent calls to set() with notify=True will trigger callbacks.

Examples:

counter = Observable[int](0)
counter.on_change(lambda value: print(f"Counter changed to {value}"))

# Disable notifications
counter.disable()
counter.set(1)  # No output

# Enable notifications
counter.enable()
counter.set(2)  # Prints: "Counter changed to 2"
Source code in observant\observable.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@override
def enable(self) -> None:
    """
    Enable the observable to notify changes.

    After calling this method, subsequent calls to set() with notify=True
    will trigger callbacks.

    Examples:
        ```python
        counter = Observable[int](0)
        counter.on_change(lambda value: print(f"Counter changed to {value}"))

        # Disable notifications
        counter.disable()
        counter.set(1)  # No output

        # Enable notifications
        counter.enable()
        counter.set(2)  # Prints: "Counter changed to 2"
        ```
    """
    print("DEBUG: Observable.enable called")
    self._on_change_enabled = True

get()

Get the current value of the observable.

Returns:

Type Description
T

The current value stored in this observable.

Examples:

counter = Observable[int](0)
value = counter.get()  # Returns: 0
Source code in observant\observable.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@override
def get(self) -> T:
    """
    Get the current value of the observable.

    Returns:
        The current value stored in this observable.

    Examples:
        ```python
        counter = Observable[int](0)
        value = counter.get()  # Returns: 0
        ```
    """
    return self._value

on_change(callback)

Register a callback function to be called when the value changes.

The callback will be called with the new value whenever set() is called with notify=True and callbacks are enabled. Callbacks are called in the order they were registered.

If the same callback function is registered multiple times, it will only be added once.

Parameters:

Name Type Description Default
callback Callable[[T], None]

A function that takes the new value as its argument.

 required

Examples:

counter = Observable[int](0)

# Register a callback
counter.on_change(lambda value: print(f"Counter changed to {value}"))

# Register another callback
counter.on_change(lambda value: print(f"Counter is now {value}"))

# Update the value
counter.set(1)
# Prints:
# "Counter changed to 1"
# "Counter is now 1"
Source code in observant\observable.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@override
def on_change(self, callback: Callable[[T], None]) -> None:
    """
    Register a callback function to be called when the value changes.

    The callback will be called with the new value whenever set() is called
    with notify=True and callbacks are enabled. Callbacks are called in the
    order they were registered.

    If the same callback function is registered multiple times, it will only
    be added once.

    Args:
        callback: A function that takes the new value as its argument.

    Examples:
        ```python
        counter = Observable[int](0)

        # Register a callback
        counter.on_change(lambda value: print(f"Counter changed to {value}"))

        # Register another callback
        counter.on_change(lambda value: print(f"Counter is now {value}"))

        # Update the value
        counter.set(1)
        # Prints:
        # "Counter changed to 1"
        # "Counter is now 1"
        ```
    """
    print(f"DEBUG: Observable.on_change called, current callbacks: {len(self._callbacks)}")
    # Check if this callback is already registered to avoid duplicates
    for existing_cb in self._callbacks:
        if existing_cb == callback:
            print("DEBUG: Observable.on_change - Callback already registered, skipping")
            return

    self._callbacks.append(callback)
    print(f"DEBUG: Observable.on_change - Added callback, now have {len(self._callbacks)} callbacks")

set(value, notify=True)

Set a new value for the observable and notify all registered callbacks.

This method updates the internal value and, if notify is True and callbacks are enabled, calls all registered callbacks with the new value.

Parameters:

Name Type Description Default
value T

The new value to set.

 required
notify bool

Whether to notify the callbacks after setting the value.

 True

Examples:

counter = Observable[int](0)
counter.on_change(lambda value: print(f"Counter changed to {value}"))

# Update with notification
counter.set(1)  # Prints: "Counter changed to 1"

# Update without notification
counter.set(2, notify=False)  # No output
Source code in observant\observable.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@override
def set(self, value: T, notify: bool = True) -> None:
    """
    Set a new value for the observable and notify all registered callbacks.

    This method updates the internal value and, if notify is True and callbacks
    are enabled, calls all registered callbacks with the new value.

    Args:
        value: The new value to set.
        notify: Whether to notify the callbacks after setting the value.

    Examples:
        ```python
        counter = Observable[int](0)
        counter.on_change(lambda value: print(f"Counter changed to {value}"))

        # Update with notification
        counter.set(1)  # Prints: "Counter changed to 1"

        # Update without notification
        counter.set(2, notify=False)  # No output
        ```
    """
    print(f"DEBUG: Observable.set called with value {value}")
    self._value = value

    if not notify or not self._on_change_enabled:
        print("DEBUG: Observable.set - on_change is disabled, skipping callbacks")
        return

    print(f"DEBUG: Observable.set - Notifying {len(self._callbacks)} callbacks")
    for i, callback in enumerate(self._callbacks):
        print(f"DEBUG: Observable.set - Calling callback {i}")
        callback(value)
        print(f"DEBUG: Observable.set - Callback {i} completed")
    print("DEBUG: Observable.set - Completed")