Skip to content

Interfaces

observant.interfaces.observable.IObservable

Bases: Generic[T]

Source code in observant\interfaces\observable.py 
 6
 7
 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
class IObservable(Generic[T]):
    def get(self) -> T:
        """
        Get the current value of the observable.

        Returns:
            The current value.
        """
        ...

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

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

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

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

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

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

disable()

Disable the observable from notifying changes.

Source code in observant\interfaces\observable.py 
41
42
43
44
45
def disable(self) -> None:
    """
    Disable the observable from notifying changes.
    """
    ...

enable()

Enable the observable to notify changes.

Source code in observant\interfaces\observable.py 
35
36
37
38
39
def enable(self) -> None:
    """
    Enable the observable to notify changes.
    """
    ...

get()

Get the current value of the observable.

Returns:

Type Description
T

The current value.
Source code in observant\interfaces\observable.py 
 7
 8
 9
10
11
12
13
14
def get(self) -> T:
    """
    Get the current value of the observable.

    Returns:
        The current value.
    """
    ...

on_change(callback)

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

Parameters:

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

A function that takes the new value as its argument.

 required
Source code in observant\interfaces\observable.py 
26
27
28
29
30
31
32
33
def on_change(self, callback: Callable[[T], None]) -> None:
    """
    Register a callback function to be called when the value changes.

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

set(value, notify=True)

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

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
Source code in observant\interfaces\observable.py 
16
17
18
19
20
21
22
23
24
def set(self, value: T, notify: bool = True) -> None:
    """
    Set a new value for the observable and notify all registered callbacks.

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

observant.interfaces.list.IObservableList

Bases: Generic[T], ABC

Interface for observable lists with specific event types.

Source code in observant\interfaces\list.py 
  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
class IObservableList(Generic[T], ABC):
    """Interface for observable lists with specific event types."""

    @abstractmethod
    def __len__(self) -> int:
        """Return the number of items in the list."""
        ...

    @abstractmethod
    def __getitem__(self, index: int | slice) -> T | list[T]:
        """Get an item or slice of items from the list."""
        ...

    @abstractmethod
    def __setitem__(self, index: int | slice, value: T | list[T]) -> None:
        """Set an item or slice of items in the list."""
        ...

    @abstractmethod
    def __delitem__(self, index: int | slice) -> None:
        """Delete an item or slice of items from the list."""
        ...

    @abstractmethod
    def __iter__(self) -> Iterator[T]:
        """Return an iterator over the items in the list."""
        ...

    @abstractmethod
    def __contains__(self, item: T) -> bool:
        """Check if an item is in the list."""
        ...

    @abstractmethod
    def append(self, item: T) -> None:
        """Add an item to the end of the list."""
        ...

    @abstractmethod
    def extend(self, items: list[T]) -> None:
        """Extend the list by appending all items from the iterable."""
        ...

    @abstractmethod
    def insert(self, index: int, item: T) -> None:
        """Insert an item at a given position."""
        ...

    @abstractmethod
    def remove(self, item: T) -> None:
        """Remove the first occurrence of an item from the list."""
        ...

    @abstractmethod
    def pop(self, index: int = -1) -> T:
        """Remove and return an item at a given position."""
        ...

    @abstractmethod
    def clear(self) -> None:
        """Remove all items from the list."""
        ...

    @abstractmethod
    def index(self, item: T, start: int = 0, end: int | None = None) -> int:
        """Return the index of the first occurrence of an item."""
        ...

    @abstractmethod
    def count(self, item: T) -> int:
        """Return the number of occurrences of an item in the list."""
        ...

    @overload
    def sort(self, *, key: None = None, reverse: bool = False) -> None: ...

    @overload
    def sort(self, *, key: Callable[[T], Any], reverse: bool = False) -> None: ...

    @abstractmethod
    def sort(
        self,
        *,
        key: Callable[[T], Any] | None = None,
        reverse: bool = False,
    ) -> None: ...

    @abstractmethod
    def reverse(self) -> None:
        """Reverse the list in place."""
        ...

    @abstractmethod
    def copy(self) -> list[T]:
        """Return a shallow copy of the list."""
        ...

    @abstractmethod
    def on_change(self, callback: Callable[[ObservableListChange[T]], None]) -> None:
        """Register for all change events with detailed information."""
        ...

    @abstractmethod
    def on_add(self, callback: Callable[[T, int], None]) -> None:
        """Register for add events with item and index."""
        ...

    @abstractmethod
    def on_remove(self, callback: Callable[[T, int], None]) -> None:
        """Register for remove events with item and index."""
        ...

    @abstractmethod
    def on_clear(self, callback: Callable[[list[T]], None]) -> None:
        """Register for clear events with the cleared items."""
        ...

__contains__(item) abstractmethod

Check if an item is in the list.

Source code in observant\interfaces\list.py 
37
38
39
40
@abstractmethod
def __contains__(self, item: T) -> bool:
    """Check if an item is in the list."""
    ...

__delitem__(index) abstractmethod

Delete an item or slice of items from the list.

Source code in observant\interfaces\list.py 
27
28
29
30
@abstractmethod
def __delitem__(self, index: int | slice) -> None:
    """Delete an item or slice of items from the list."""
    ...

__getitem__(index) abstractmethod

Get an item or slice of items from the list.

Source code in observant\interfaces\list.py 
17
18
19
20
@abstractmethod
def __getitem__(self, index: int | slice) -> T | list[T]:
    """Get an item or slice of items from the list."""
    ...

__iter__() abstractmethod

Return an iterator over the items in the list.

Source code in observant\interfaces\list.py 
32
33
34
35
@abstractmethod
def __iter__(self) -> Iterator[T]:
    """Return an iterator over the items in the list."""
    ...

__len__() abstractmethod

Return the number of items in the list.

Source code in observant\interfaces\list.py 
12
13
14
15
@abstractmethod
def __len__(self) -> int:
    """Return the number of items in the list."""
    ...

__setitem__(index, value) abstractmethod

Set an item or slice of items in the list.

Source code in observant\interfaces\list.py 
22
23
24
25
@abstractmethod
def __setitem__(self, index: int | slice, value: T | list[T]) -> None:
    """Set an item or slice of items in the list."""
    ...

append(item) abstractmethod

Add an item to the end of the list.

Source code in observant\interfaces\list.py 
42
43
44
45
@abstractmethod
def append(self, item: T) -> None:
    """Add an item to the end of the list."""
    ...

clear() abstractmethod

Remove all items from the list.

Source code in observant\interfaces\list.py 
67
68
69
70
@abstractmethod
def clear(self) -> None:
    """Remove all items from the list."""
    ...

copy() abstractmethod

Return a shallow copy of the list.

Source code in observant\interfaces\list.py 
101
102
103
104
@abstractmethod
def copy(self) -> list[T]:
    """Return a shallow copy of the list."""
    ...

count(item) abstractmethod

Return the number of occurrences of an item in the list.

Source code in observant\interfaces\list.py 
77
78
79
80
@abstractmethod
def count(self, item: T) -> int:
    """Return the number of occurrences of an item in the list."""
    ...

extend(items) abstractmethod

Extend the list by appending all items from the iterable.

Source code in observant\interfaces\list.py 
47
48
49
50
@abstractmethod
def extend(self, items: list[T]) -> None:
    """Extend the list by appending all items from the iterable."""
    ...

index(item, start=0, end=None) abstractmethod

Return the index of the first occurrence of an item.

Source code in observant\interfaces\list.py 
72
73
74
75
@abstractmethod
def index(self, item: T, start: int = 0, end: int | None = None) -> int:
    """Return the index of the first occurrence of an item."""
    ...

insert(index, item) abstractmethod

Insert an item at a given position.

Source code in observant\interfaces\list.py 
52
53
54
55
@abstractmethod
def insert(self, index: int, item: T) -> None:
    """Insert an item at a given position."""
    ...

on_add(callback) abstractmethod

Register for add events with item and index.

Source code in observant\interfaces\list.py 
111
112
113
114
@abstractmethod
def on_add(self, callback: Callable[[T, int], None]) -> None:
    """Register for add events with item and index."""
    ...

on_change(callback) abstractmethod

Register for all change events with detailed information.

Source code in observant\interfaces\list.py 
106
107
108
109
@abstractmethod
def on_change(self, callback: Callable[[ObservableListChange[T]], None]) -> None:
    """Register for all change events with detailed information."""
    ...

on_clear(callback) abstractmethod

Register for clear events with the cleared items.

Source code in observant\interfaces\list.py 
121
122
123
124
@abstractmethod
def on_clear(self, callback: Callable[[list[T]], None]) -> None:
    """Register for clear events with the cleared items."""
    ...

on_remove(callback) abstractmethod

Register for remove events with item and index.

Source code in observant\interfaces\list.py 
116
117
118
119
@abstractmethod
def on_remove(self, callback: Callable[[T, int], None]) -> None:
    """Register for remove events with item and index."""
    ...

pop(index=-1) abstractmethod

Remove and return an item at a given position.

Source code in observant\interfaces\list.py 
62
63
64
65
@abstractmethod
def pop(self, index: int = -1) -> T:
    """Remove and return an item at a given position."""
    ...

remove(item) abstractmethod

Remove the first occurrence of an item from the list.

Source code in observant\interfaces\list.py 
57
58
59
60
@abstractmethod
def remove(self, item: T) -> None:
    """Remove the first occurrence of an item from the list."""
    ...

reverse() abstractmethod

Reverse the list in place.

Source code in observant\interfaces\list.py 
96
97
98
99
@abstractmethod
def reverse(self) -> None:
    """Reverse the list in place."""
    ...

observant.interfaces.dict.IObservableDict

Bases: Generic[TKey, TValue], ABC

Interface for observable dictionaries with specific event types.

Source code in observant\interfaces\dict.py 
 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
class IObservableDict(Generic[TKey, TValue], ABC):
    """Interface for observable dictionaries with specific event types."""

    @abstractmethod
    def __len__(self) -> int:
        """Return the number of items in the dictionary."""
        ...

    @abstractmethod
    def __getitem__(self, key: TKey) -> TValue:
        """Get an item from the dictionary."""
        ...

    @abstractmethod
    def __setitem__(self, key: TKey, value: TValue) -> None:
        """Set an item in the dictionary."""
        ...

    @abstractmethod
    def __delitem__(self, key: TKey) -> None:
        """Delete an item from the dictionary."""
        ...

    @abstractmethod
    def __iter__(self) -> Iterator[TKey]:
        """Return an iterator over the keys in the dictionary."""
        ...

    @abstractmethod
    def __contains__(self, key: TKey) -> bool:
        """Check if a key is in the dictionary."""
        ...

    @abstractmethod
    def get(self, key: TKey, default: TValue | None = None) -> TValue | None:
        """Return the value for a key if it exists, otherwise return a default value."""
        ...

    @abstractmethod
    def setdefault(self, key: TKey, default: TValue | None = None) -> TValue | None:
        """Return the value for a key if it exists, otherwise set and return the default value."""
        ...

    @abstractmethod
    def pop(self, key: TKey, default: TValue | None = None) -> TValue | None:
        """Remove and return the value for a key if it exists, otherwise return a default value."""
        ...

    @abstractmethod
    def popitem(self) -> tuple[TKey, TValue]:
        """Remove and return a (key, value) pair from the dictionary."""
        ...

    @abstractmethod
    def clear(self) -> None:
        """Remove all items from the dictionary."""
        ...

    @abstractmethod
    def update(self, other: dict[TKey, TValue]) -> None:
        """Update the dictionary with the key/value pairs from another dictionary."""
        ...

    @abstractmethod
    def keys(self) -> list[TKey]:
        """Return a list of all keys in the dictionary."""
        ...

    @abstractmethod
    def values(self) -> list[TValue]:
        """Return a list of all values in the dictionary."""
        ...

    @abstractmethod
    def items(self) -> list[tuple[TKey, TValue]]:
        """Return a list of all (key, value) pairs in the dictionary."""
        ...

    @abstractmethod
    def copy(self) -> dict[TKey, TValue]:
        """Return a shallow copy of the dictionary."""
        ...

    @abstractmethod
    def on_change(
        self, callback: Callable[[ObservableDictChange[TKey, TValue]], None]
    ) -> None:
        """Register for all change events with detailed information."""
        ...

    @abstractmethod
    def on_add(self, callback: Callable[[TKey, TValue], None]) -> None:
        """Register for add events with key and value."""
        ...

    @abstractmethod
    def on_remove(self, callback: Callable[[TKey, TValue], None]) -> None:
        """Register for remove events with key and value."""
        ...

    @abstractmethod
    def on_update(self, callback: Callable[[TKey, TValue], None]) -> None:
        """Register for update events with key and new value."""
        ...

    @abstractmethod
    def on_clear(self, callback: Callable[[dict[TKey, TValue]], None]) -> None:
        """Register for clear events with the cleared items."""
        ...

__contains__(key) abstractmethod

Check if a key is in the dictionary.

Source code in observant\interfaces\dict.py 
38
39
40
41
@abstractmethod
def __contains__(self, key: TKey) -> bool:
    """Check if a key is in the dictionary."""
    ...

__delitem__(key) abstractmethod

Delete an item from the dictionary.

Source code in observant\interfaces\dict.py 
28
29
30
31
@abstractmethod
def __delitem__(self, key: TKey) -> None:
    """Delete an item from the dictionary."""
    ...

__getitem__(key) abstractmethod

Get an item from the dictionary.

Source code in observant\interfaces\dict.py 
18
19
20
21
@abstractmethod
def __getitem__(self, key: TKey) -> TValue:
    """Get an item from the dictionary."""
    ...

__iter__() abstractmethod

Return an iterator over the keys in the dictionary.

Source code in observant\interfaces\dict.py 
33
34
35
36
@abstractmethod
def __iter__(self) -> Iterator[TKey]:
    """Return an iterator over the keys in the dictionary."""
    ...

__len__() abstractmethod

Return the number of items in the dictionary.

Source code in observant\interfaces\dict.py 
13
14
15
16
@abstractmethod
def __len__(self) -> int:
    """Return the number of items in the dictionary."""
    ...

__setitem__(key, value) abstractmethod

Set an item in the dictionary.

Source code in observant\interfaces\dict.py 
23
24
25
26
@abstractmethod
def __setitem__(self, key: TKey, value: TValue) -> None:
    """Set an item in the dictionary."""
    ...

clear() abstractmethod

Remove all items from the dictionary.

Source code in observant\interfaces\dict.py 
63
64
65
66
@abstractmethod
def clear(self) -> None:
    """Remove all items from the dictionary."""
    ...

copy() abstractmethod

Return a shallow copy of the dictionary.

Source code in observant\interfaces\dict.py 
88
89
90
91
@abstractmethod
def copy(self) -> dict[TKey, TValue]:
    """Return a shallow copy of the dictionary."""
    ...

get(key, default=None) abstractmethod

Return the value for a key if it exists, otherwise return a default value.

Source code in observant\interfaces\dict.py 
43
44
45
46
@abstractmethod
def get(self, key: TKey, default: TValue | None = None) -> TValue | None:
    """Return the value for a key if it exists, otherwise return a default value."""
    ...

items() abstractmethod

Return a list of all (key, value) pairs in the dictionary.

Source code in observant\interfaces\dict.py 
83
84
85
86
@abstractmethod
def items(self) -> list[tuple[TKey, TValue]]:
    """Return a list of all (key, value) pairs in the dictionary."""
    ...

keys() abstractmethod

Return a list of all keys in the dictionary.

Source code in observant\interfaces\dict.py 
73
74
75
76
@abstractmethod
def keys(self) -> list[TKey]:
    """Return a list of all keys in the dictionary."""
    ...

on_add(callback) abstractmethod

Register for add events with key and value.

Source code in observant\interfaces\dict.py 
100
101
102
103
@abstractmethod
def on_add(self, callback: Callable[[TKey, TValue], None]) -> None:
    """Register for add events with key and value."""
    ...

on_change(callback) abstractmethod

Register for all change events with detailed information.

Source code in observant\interfaces\dict.py 
93
94
95
96
97
98
@abstractmethod
def on_change(
    self, callback: Callable[[ObservableDictChange[TKey, TValue]], None]
) -> None:
    """Register for all change events with detailed information."""
    ...

on_clear(callback) abstractmethod

Register for clear events with the cleared items.

Source code in observant\interfaces\dict.py 
115
116
117
118
@abstractmethod
def on_clear(self, callback: Callable[[dict[TKey, TValue]], None]) -> None:
    """Register for clear events with the cleared items."""
    ...

on_remove(callback) abstractmethod

Register for remove events with key and value.

Source code in observant\interfaces\dict.py 
105
106
107
108
@abstractmethod
def on_remove(self, callback: Callable[[TKey, TValue], None]) -> None:
    """Register for remove events with key and value."""
    ...

on_update(callback) abstractmethod

Register for update events with key and new value.

Source code in observant\interfaces\dict.py 
110
111
112
113
@abstractmethod
def on_update(self, callback: Callable[[TKey, TValue], None]) -> None:
    """Register for update events with key and new value."""
    ...

pop(key, default=None) abstractmethod

Remove and return the value for a key if it exists, otherwise return a default value.

Source code in observant\interfaces\dict.py 
53
54
55
56
@abstractmethod
def pop(self, key: TKey, default: TValue | None = None) -> TValue | None:
    """Remove and return the value for a key if it exists, otherwise return a default value."""
    ...

popitem() abstractmethod

Remove and return a (key, value) pair from the dictionary.

Source code in observant\interfaces\dict.py 
58
59
60
61
@abstractmethod
def popitem(self) -> tuple[TKey, TValue]:
    """Remove and return a (key, value) pair from the dictionary."""
    ...

setdefault(key, default=None) abstractmethod

Return the value for a key if it exists, otherwise set and return the default value.

Source code in observant\interfaces\dict.py 
48
49
50
51
@abstractmethod
def setdefault(self, key: TKey, default: TValue | None = None) -> TValue | None:
    """Return the value for a key if it exists, otherwise set and return the default value."""
    ...

update(other) abstractmethod

Update the dictionary with the key/value pairs from another dictionary.

Source code in observant\interfaces\dict.py 
68
69
70
71
@abstractmethod
def update(self, other: dict[TKey, TValue]) -> None:
    """Update the dictionary with the key/value pairs from another dictionary."""
    ...

values() abstractmethod

Return a list of all values in the dictionary.

Source code in observant\interfaces\dict.py 
78
79
80
81
@abstractmethod
def values(self) -> list[TValue]:
    """Return a list of all values in the dictionary."""
    ...

observant.interfaces.proxy.IObservableProxy

Bases: Generic[T], ABC

Proxy for a data object that exposes its fields as Observable, ObservableList, or ObservableDict. Provides optional sync behavior to automatically write back to the source model.

Source code in observant\interfaces\proxy.py 
 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
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
class IObservableProxy(Generic[T], ABC):
    """
    Proxy for a data object that exposes its fields as Observable, ObservableList, or ObservableDict.
    Provides optional sync behavior to automatically write back to the source model.
    """

    @abstractmethod
    def observable(
        self,
        typ: type[T],
        attr: str,
        *,
        sync: bool | None = None,
    ) -> IObservable[T]:
        """
        Get or create an Observable[T] for a scalar field.
        """
        ...

    @abstractmethod
    def observable_list(
        self,
        typ: type[T],
        attr: str,
        *,
        sync: bool | None = None,
    ) -> IObservableList[T]:
        """
        Get or create an ObservableList[T] for a list field.
        """
        ...

    @abstractmethod
    def observable_dict(
        self,
        typ: tuple[type[TKey], type[TValue]],
        attr: str,
        *,
        sync: bool | None = None,
    ) -> IObservableDict[TKey, TValue]:
        """
        Get or create an ObservableDict for a dict field.
        """
        ...

    @abstractmethod
    def get(self) -> T:
        """
        Get the original object being proxied.
        """
        ...

    @abstractmethod
    def update(self, **kwargs: Any) -> None:
        """
        Set one or more scalar observable values.
        """
        ...

    @abstractmethod
    def load_dict(self, values: dict[str, Any]) -> None:
        """
        Set multiple scalar observable values from a dict.
        """
        ...

    @abstractmethod
    def save_to(self, obj: T) -> None:
        """
        Write all observable values back into the given object.
        """
        ...

    @abstractmethod
    def is_dirty(self) -> bool:
        """
        Check if any fields have been modified since initialization or last reset.

        Returns:
            True if any fields have been modified, False otherwise.
        """
        ...

    @abstractmethod
    def dirty_fields(self) -> set[str]:
        """
        Get the set of field names that have been modified.

        Returns:
            A set of field names that have been modified.
        """
        ...

    @abstractmethod
    def reset_dirty(self) -> None:
        """
        Reset the dirty state of all fields.
        """
        ...

    @abstractmethod
    def register_computed(
        self,
        name: str,
        compute: Callable[[], T],
        dependencies: list[str],
    ) -> None:
        """
        Register a computed property that depends on other observables.

        Args:
            name: The name of the computed property.
            compute: A function that returns the computed value.
            dependencies: List of field names that this computed property depends on.
        """
        ...

    @abstractmethod
    def computed(
        self,
        typ: type[T],
        name: str,
    ) -> IObservable[T]:
        """
        Get a computed property by name.

        Args:
            typ: The type of the computed property.
            name: The name of the computed property.

        Returns:
            An observable containing the computed value.
        """
        ...

    @abstractmethod
    def add_validator(
        self,
        attr: str,
        validator: Callable[[Any], str | None],
    ) -> None:
        """
        Add a validator function for a field.

        Args:
            attr: The field name to validate.
            validator: A function that takes the field value and returns an error message
                       if invalid, or None if valid.
        """
        ...

    @abstractmethod
    def is_valid(self) -> IObservable[bool]:
        """
        Get an observable that indicates whether all fields are valid.

        Returns:
            An observable that emits True if all fields are valid, False otherwise.
        """
        ...

    @abstractmethod
    def validation_errors(self) -> IObservableDict[str, list[str]]:
        """
        Get an observable dictionary of validation errors.

        Returns:
            An observable dictionary mapping field names to lists of error messages.
        """
        ...

    @abstractmethod
    def validation_for(self, attr: str) -> IObservable[list[str]]:
        """
        Get an observable list of validation errors for a specific field.

        Args:
            attr: The field name to get validation errors for.

        Returns:
            An observable that emits a list of error messages for the field.
            An empty list means the field is valid.
        """
        ...

    @abstractmethod
    def reset_validation(self, attr: str | None = None, *, revalidate: bool = False) -> None:
        """
        Reset validation errors for a specific field or all fields.

        Args:
            attr: The field name to reset validation for. If None, reset all fields.
            revalidate: Whether to re-run validators after clearing errors.
        """
        ...

    @abstractmethod
    def set_undo_config(
        self,
        attr: str,
        *,
        undo_max: int | None = None,
        undo_debounce_ms: int | None = None,
    ) -> None:
        """
        Set the undo configuration for a specific field.

        Args:
            attr: The field name to configure.
            undo_max: Maximum number of undo steps to store. None means unlimited.
            undo_debounce_ms: Time window in milliseconds to group changes. None means no debouncing.
        """
        ...

    @abstractmethod
    def undo(self, attr: str) -> None:
        """
        Undo the most recent change to a field.

        Args:
            attr: The field name to undo changes for.
        """
        ...

    @abstractmethod
    def redo(self, attr: str) -> None:
        """
        Redo the most recently undone change to a field.

        Args:
            attr: The field name to redo changes for.
        """
        ...

    @abstractmethod
    def can_undo(self, attr: str) -> bool:
        """
        Check if there are changes that can be undone for a field.

        Args:
            attr: The field name to check.

        Returns:
            True if there are changes that can be undone, False otherwise.
        """
        ...

    @abstractmethod
    def can_redo(self, attr: str) -> bool:
        """
        Check if there are changes that can be redone for a field.

        Args:
            attr: The field name to check.

        Returns:
            True if there are changes that can be redone, False otherwise.
        """
        ...

    @abstractmethod
    def track_scalar_change(self, attr: str, old_value: Any, new_value: Any) -> None:
        """
        Track a scalar change for undo/redo functionality.

        Args:
            attr: The field name that changed.
            old_value: The old value before the change.
            new_value: The new value after the change.
        """
        ...

add_validator(attr, validator) abstractmethod

Add a validator function for a field.

Parameters:

Name Type Description Default
attr str

The field name to validate.

 required
validator Callable[[Any], str | None]

A function that takes the field value and returns an error message if invalid, or None if valid.

 required
Source code in observant\interfaces\proxy.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@abstractmethod
def add_validator(
    self,
    attr: str,
    validator: Callable[[Any], str | None],
) -> None:
    """
    Add a validator function for a field.

    Args:
        attr: The field name to validate.
        validator: A function that takes the field value and returns an error message
                   if invalid, or None if valid.
    """
    ...

can_redo(attr) abstractmethod

Check if there are changes that can be redone for a field.

Parameters:

Name Type Description Default
attr str

The field name to check.

 required

Returns:

Type Description
bool

True if there are changes that can be redone, False otherwise.
Source code in observant\interfaces\proxy.py 
260
261
262
263
264
265
266
267
268
269
270
271
@abstractmethod
def can_redo(self, attr: str) -> bool:
    """
    Check if there are changes that can be redone for a field.

    Args:
        attr: The field name to check.

    Returns:
        True if there are changes that can be redone, False otherwise.
    """
    ...

can_undo(attr) abstractmethod

Check if there are changes that can be undone for a field.

Parameters:

Name Type Description Default
attr str

The field name to check.

 required

Returns:

Type Description
bool

True if there are changes that can be undone, False otherwise.
Source code in observant\interfaces\proxy.py 
247
248
249
250
251
252
253
254
255
256
257
258
@abstractmethod
def can_undo(self, attr: str) -> bool:
    """
    Check if there are changes that can be undone for a field.

    Args:
        attr: The field name to check.

    Returns:
        True if there are changes that can be undone, False otherwise.
    """
    ...

computed(typ, name) abstractmethod

Get a computed property by name.

Parameters:

Name Type Description Default
typ type[T]

The type of the computed property.

 required
name str

The name of the computed property.

 required

Returns:

Type Description
IObservable[T]

An observable containing the computed value.
Source code in observant\interfaces\proxy.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@abstractmethod
def computed(
    self,
    typ: type[T],
    name: str,
) -> IObservable[T]:
    """
    Get a computed property by name.

    Args:
        typ: The type of the computed property.
        name: The name of the computed property.

    Returns:
        An observable containing the computed value.
    """
    ...

dirty_fields() abstractmethod

Get the set of field names that have been modified.

Returns:

Type Description
set[str]

A set of field names that have been modified.
Source code in observant\interfaces\proxy.py 
 96
 97
 98
 99
100
101
102
103
104
@abstractmethod
def dirty_fields(self) -> set[str]:
    """
    Get the set of field names that have been modified.

    Returns:
        A set of field names that have been modified.
    """
    ...

get() abstractmethod

Get the original object being proxied.

Source code in observant\interfaces\proxy.py 
58
59
60
61
62
63
@abstractmethod
def get(self) -> T:
    """
    Get the original object being proxied.
    """
    ...

is_dirty() abstractmethod

Check if any fields have been modified since initialization or last reset.

Returns:

Type Description
bool

True if any fields have been modified, False otherwise.
Source code in observant\interfaces\proxy.py 
86
87
88
89
90
91
92
93
94
@abstractmethod
def is_dirty(self) -> bool:
    """
    Check if any fields have been modified since initialization or last reset.

    Returns:
        True if any fields have been modified, False otherwise.
    """
    ...

is_valid() abstractmethod

Get an observable that indicates whether all fields are valid.

Returns:

Type Description
IObservable[bool]

An observable that emits True if all fields are valid, False otherwise.
Source code in observant\interfaces\proxy.py 
164
165
166
167
168
169
170
171
172
@abstractmethod
def is_valid(self) -> IObservable[bool]:
    """
    Get an observable that indicates whether all fields are valid.

    Returns:
        An observable that emits True if all fields are valid, False otherwise.
    """
    ...

load_dict(values) abstractmethod

Set multiple scalar observable values from a dict.

Source code in observant\interfaces\proxy.py 
72
73
74
75
76
77
@abstractmethod
def load_dict(self, values: dict[str, Any]) -> None:
    """
    Set multiple scalar observable values from a dict.
    """
    ...

observable(typ, attr, *, sync=None) abstractmethod

Get or create an Observable[T] for a scalar field.

Source code in observant\interfaces\proxy.py 
19
20
21
22
23
24
25
26
27
28
29
30
@abstractmethod
def observable(
    self,
    typ: type[T],
    attr: str,
    *,
    sync: bool | None = None,
) -> IObservable[T]:
    """
    Get or create an Observable[T] for a scalar field.
    """
    ...

observable_dict(typ, attr, *, sync=None) abstractmethod

Get or create an ObservableDict for a dict field.

Source code in observant\interfaces\proxy.py 
45
46
47
48
49
50
51
52
53
54
55
56
@abstractmethod
def observable_dict(
    self,
    typ: tuple[type[TKey], type[TValue]],
    attr: str,
    *,
    sync: bool | None = None,
) -> IObservableDict[TKey, TValue]:
    """
    Get or create an ObservableDict for a dict field.
    """
    ...

observable_list(typ, attr, *, sync=None) abstractmethod

Get or create an ObservableList[T] for a list field.

Source code in observant\interfaces\proxy.py 
32
33
34
35
36
37
38
39
40
41
42
43
@abstractmethod
def observable_list(
    self,
    typ: type[T],
    attr: str,
    *,
    sync: bool | None = None,
) -> IObservableList[T]:
    """
    Get or create an ObservableList[T] for a list field.
    """
    ...

redo(attr) abstractmethod

Redo the most recently undone change to a field.

Parameters:

Name Type Description Default
attr str

The field name to redo changes for.

 required
Source code in observant\interfaces\proxy.py 
237
238
239
240
241
242
243
244
245
@abstractmethod
def redo(self, attr: str) -> None:
    """
    Redo the most recently undone change to a field.

    Args:
        attr: The field name to redo changes for.
    """
    ...

register_computed(name, compute, dependencies) abstractmethod

Register a computed property that depends on other observables.

Parameters:

Name Type Description Default
name str

The name of the computed property.

 required
compute Callable[[], T]

A function that returns the computed value.

 required
dependencies list[str]

List of field names that this computed property depends on.

 required
Source code in observant\interfaces\proxy.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@abstractmethod
def register_computed(
    self,
    name: str,
    compute: Callable[[], T],
    dependencies: list[str],
) -> None:
    """
    Register a computed property that depends on other observables.

    Args:
        name: The name of the computed property.
        compute: A function that returns the computed value.
        dependencies: List of field names that this computed property depends on.
    """
    ...

reset_dirty() abstractmethod

Reset the dirty state of all fields.

Source code in observant\interfaces\proxy.py 
106
107
108
109
110
111
@abstractmethod
def reset_dirty(self) -> None:
    """
    Reset the dirty state of all fields.
    """
    ...

reset_validation(attr=None, *, revalidate=False) abstractmethod

Reset validation errors for a specific field or all fields.

Parameters:

Name Type Description Default
attr str | None

The field name to reset validation for. If None, reset all fields.

 None
revalidate bool

Whether to re-run validators after clearing errors.

 False
Source code in observant\interfaces\proxy.py 
198
199
200
201
202
203
204
205
206
207
@abstractmethod
def reset_validation(self, attr: str | None = None, *, revalidate: bool = False) -> None:
    """
    Reset validation errors for a specific field or all fields.

    Args:
        attr: The field name to reset validation for. If None, reset all fields.
        revalidate: Whether to re-run validators after clearing errors.
    """
    ...

save_to(obj) abstractmethod

Write all observable values back into the given object.

Source code in observant\interfaces\proxy.py 
79
80
81
82
83
84
@abstractmethod
def save_to(self, obj: T) -> None:
    """
    Write all observable values back into the given object.
    """
    ...

set_undo_config(attr, *, undo_max=None, undo_debounce_ms=None) abstractmethod

Set the undo configuration for a specific field.

Parameters:

Name Type Description Default
attr str

The field name to configure.

 required
undo_max int | None

Maximum number of undo steps to store. None means unlimited.

 None
undo_debounce_ms int | None

Time window in milliseconds to group changes. None means no debouncing.

 None
Source code in observant\interfaces\proxy.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@abstractmethod
def set_undo_config(
    self,
    attr: str,
    *,
    undo_max: int | None = None,
    undo_debounce_ms: int | None = None,
) -> None:
    """
    Set the undo configuration for a specific field.

    Args:
        attr: The field name to configure.
        undo_max: Maximum number of undo steps to store. None means unlimited.
        undo_debounce_ms: Time window in milliseconds to group changes. None means no debouncing.
    """
    ...

track_scalar_change(attr, old_value, new_value) abstractmethod

Track a scalar change for undo/redo functionality.

Parameters:

Name Type Description Default
attr str

The field name that changed.

 required
old_value Any

The old value before the change.

 required
new_value Any

The new value after the change.

 required
Source code in observant\interfaces\proxy.py 
273
274
275
276
277
278
279
280
281
282
283
@abstractmethod
def track_scalar_change(self, attr: str, old_value: Any, new_value: Any) -> None:
    """
    Track a scalar change for undo/redo functionality.

    Args:
        attr: The field name that changed.
        old_value: The old value before the change.
        new_value: The new value after the change.
    """
    ...

undo(attr) abstractmethod

Undo the most recent change to a field.

Parameters:

Name Type Description Default
attr str

The field name to undo changes for.

 required
Source code in observant\interfaces\proxy.py 
227
228
229
230
231
232
233
234
235
@abstractmethod
def undo(self, attr: str) -> None:
    """
    Undo the most recent change to a field.

    Args:
        attr: The field name to undo changes for.
    """
    ...

update(**kwargs) abstractmethod

Set one or more scalar observable values.

Source code in observant\interfaces\proxy.py 
65
66
67
68
69
70
@abstractmethod
def update(self, **kwargs: Any) -> None:
    """
    Set one or more scalar observable values.
    """
    ...

validation_errors() abstractmethod

Get an observable dictionary of validation errors.

Returns:

Type Description
IObservableDict[str, list[str]]

An observable dictionary mapping field names to lists of error messages.
Source code in observant\interfaces\proxy.py 
174
175
176
177
178
179
180
181
182
@abstractmethod
def validation_errors(self) -> IObservableDict[str, list[str]]:
    """
    Get an observable dictionary of validation errors.

    Returns:
        An observable dictionary mapping field names to lists of error messages.
    """
    ...

validation_for(attr) abstractmethod

Get an observable list of validation errors for a specific field.

Parameters:

Name Type Description Default
attr str

The field name to get validation errors for.

 required

Returns:

Type Description
IObservable[list[str]]

An observable that emits a list of error messages for the field.
IObservable[list[str]]

An empty list means the field is valid.
Source code in observant\interfaces\proxy.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
@abstractmethod
def validation_for(self, attr: str) -> IObservable[list[str]]:
    """
    Get an observable list of validation errors for a specific field.

    Args:
        attr: The field name to get validation errors for.

    Returns:
        An observable that emits a list of error messages for the field.
        An empty list means the field is valid.
    """
    ...