Skip to content

ObservableDict

observant.observable_dict.ObservableDict

Bases: Generic[TKey, TValue], IObservableDict[TKey, TValue]

An observable implementation of Python's dictionary that notifies listeners of changes.

ObservableDict wraps a Python dictionary and provides the same interface, but with additional notification capabilities. It can either create its own internal dictionary or work with an existing one.

When the dictionary is modified (items added, removed, updated, etc.), registered callbacks are notified with details about the change. This allows other components to react to changes in the dictionary.

Attributes:

Name Type Description
_items dict[TKey, TValue]

The internal dictionary being observed.

_change_callbacks list[Callable[[ObservableDictChange[TKey, TValue]], None]]

Callbacks for all types of changes.

_add_callbacks list[Callable[[TKey, TValue], None]]

Callbacks specifically for add operations.

_remove_callbacks list[Callable[[TKey, TValue], None]]

Callbacks specifically for remove operations.

_update_callbacks list[Callable[[TKey, TValue], None]]

Callbacks specifically for update operations.

_clear_callbacks list[Callable[[dict[TKey, TValue]], None]]

Callbacks specifically for clear operations.

Examples:

# Create an empty observable dictionary
settings = ObservableDict[str, int]()

# Create with initial items
user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})

# Register a callback for all changes
user_data.on_change(lambda change: print(f"Change: {change.type}"))

# Register a callback for adds
user_data.on_add(lambda key, value: print(f"Added {key}: {value}"))

# Modify the dictionary
user_data["phone"] = "555-1234"  # Triggers callbacks
Source code in observant\observable_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
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
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
class ObservableDict(Generic[TKey, TValue], IObservableDict[TKey, TValue]):
    """
    An observable implementation of Python's dictionary that notifies listeners of changes.

    ObservableDict wraps a Python dictionary and provides the same interface, but with
    additional notification capabilities. It can either create its own internal
    dictionary or work with an existing one.

    When the dictionary is modified (items added, removed, updated, etc.), registered
    callbacks are notified with details about the change. This allows other components
    to react to changes in the dictionary.

    Attributes:
        _items: The internal dictionary being observed.
        _change_callbacks: Callbacks for all types of changes.
        _add_callbacks: Callbacks specifically for add operations.
        _remove_callbacks: Callbacks specifically for remove operations.
        _update_callbacks: Callbacks specifically for update operations.
        _clear_callbacks: Callbacks specifically for clear operations.

    Examples:
        ```python
        # Create an empty observable dictionary
        settings = ObservableDict[str, int]()

        # Create with initial items
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})

        # Register a callback for all changes
        user_data.on_change(lambda change: print(f"Change: {change.type}"))

        # Register a callback for adds
        user_data.on_add(lambda key, value: print(f"Added {key}: {value}"))

        # Modify the dictionary
        user_data["phone"] = "555-1234"  # Triggers callbacks
        ```
    """

    def __init__(self, items: dict[TKey, TValue] | None = None, *, copy: bool = False) -> None:
        """
        Initialize with optional external dict reference.

        Args:
            items: Optional external dict to observe. If None, creates a new dict.
            copy: If True, creates a copy of the provided dict instead of using it directly.
                This is useful when you want to avoid modifying the original dict.

        Examples:
            ```python
            # Create an empty observable dictionary
            empty_dict = ObservableDict[str, int]()

            # Create with initial items
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})

            # Create with a copy of initial items
            original = {"red": "#FF0000", "green": "#00FF00"}
            colors = ObservableDict[str, str](original, copy=True)
            colors["blue"] = "#0000FF"  # original dict is not modified
            ```
        """
        if copy:
            self._items: dict[TKey, TValue] = dict(items) if items is not None else {}
        else:
            self._items = items if items is not None else {}
        self._change_callbacks: list[Callable[[ObservableDictChange[TKey, TValue]], None]] = []
        self._add_callbacks: list[Callable[[TKey, TValue], None]] = []
        self._remove_callbacks: list[Callable[[TKey, TValue], None]] = []
        self._update_callbacks: list[Callable[[TKey, TValue], None]] = []
        self._clear_callbacks: list[Callable[[dict[TKey, TValue]], None]] = []

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

        Returns:
            The number of items in the dictionary.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            length = len(user_data)  # Returns: 2
            ```
        """
        return len(self._items)

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

        Args:
            key: The key to look up.

        Returns:
            The value for the key.

        Raises:
            KeyError: If the key is not found.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            name = user_data["name"]  # Returns: "Alice"
            ```
        """
        return self._items[key]

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

        This method notifies callbacks about the added or updated item.
        If the key already exists, an update notification is sent.
        If the key is new, an add notification is sent.

        Args:
            key: The key to set.
            value: The value to set.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice"})
            user_data["email"] = "alice@example.com"  # Triggers add callbacks
            user_data["name"] = "Alicia"  # Triggers update callbacks
            ```
        """
        if key in self._items:
            self._items[key] = value
            self._notify_update(key, value)
        else:
            self._items[key] = value
            self._notify_add(key, value)

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

        This method notifies callbacks about the removed item.

        Args:
            key: The key to delete.

        Raises:
            KeyError: If the key is not found.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            del user_data["email"]  # Triggers remove callbacks
            ```
        """
        value = self._items[key]
        del self._items[key]
        self._notify_remove(key, value)

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

        Returns:
            An iterator over the keys in the dictionary.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            for key in user_data:
                print(key)  # Prints: "name", "email"
            ```
        """
        return iter(self._items)

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

        Args:
            key: The key to check for.

        Returns:
            True if the key is in the dictionary, False otherwise.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            if "name" in user_data:
                print("Name is present")
            ```
        """
        return key in self._items

    @override
    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.

        This method does not modify the dictionary or trigger any callbacks.

        Args:
            key: The key to look up.
            default: The default value to return if the key is not found.

        Returns:
            The value for the key, or the default value.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice"})
            email = user_data.get("email", "No email")  # Returns: "No email"
            name = user_data.get("name", "Unknown")  # Returns: "Alice"
            ```
        """
        return self._items.get(key, default)

    @override
    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.

        If the key does not exist, this method adds it with the default value and
        triggers add callbacks.

        Args:
            key: The key to look up.
            default: The default value to set and return if the key is not found.

        Returns:
            The value for the key, or the default value.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice"})
            email = user_data.setdefault("email", "alice@example.com")  # Returns: "alice@example.com" and adds it
            name = user_data.setdefault("name", "Unknown")  # Returns: "Alice" without changing it
            ```
        """
        if key not in self._items:
            self._items[key] = cast(TValue, default)  # Cast to V since we know it's a value
            self._notify_add(key, cast(TValue, default))
            return default
        return self._items[key]

    @override
    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.

        If the key exists, this method removes it and triggers remove callbacks.
        If the key does not exist and a default is provided, no callbacks are triggered.

        Args:
            key: The key to look up.
            default: The default value to return if the key is not found.

        Returns:
            The value for the key, or the default value.

        Raises:
            KeyError: If the key is not found and no default value is provided.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            email = user_data.pop("email")  # Returns: "alice@example.com" and removes it
            phone = user_data.pop("phone", "No phone")  # Returns: "No phone" without modifying the dict
            ```
        """
        if key in self._items:
            value = self._items.pop(key)
            self._notify_remove(key, value)
            return value
        if default is not None:
            return default
        raise KeyError(key)

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

        This method removes an arbitrary (key, value) pair and triggers remove callbacks.
        In Python 3.7+, the pairs are returned in LIFO order (last inserted, first returned).

        Returns:
            A (key, value) pair.

        Raises:
            KeyError: If the dictionary is empty.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            key, value = user_data.popitem()  # Might return: ("email", "alice@example.com")
            ```
        """
        key, value = self._items.popitem()
        self._notify_remove(key, value)
        return key, value

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

        This method notifies callbacks about the cleared items.
        If the dictionary is already empty, no notifications are sent.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            user_data.clear()  # Dictionary becomes {}
            ```
        """
        if not self._items:
            return
        items = self._items.copy()
        self._items.clear()
        self._notify_clear(items)

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

        This method notifies callbacks about added and updated items.
        For each key in the other dictionary:
        - If the key already exists, an update notification is sent.
        - If the key is new, an add notification is sent.
        If the other dictionary is empty, no notifications are sent.

        Args:
            other: Another dictionary to update from.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice"})
            user_data.update({"email": "alice@example.com", "name": "Alicia"})
            # Dictionary becomes {"name": "Alicia", "email": "alice@example.com"}
            ```
        """
        if not other:
            return
        added_items: dict[TKey, TValue] = {}
        updated_items: dict[TKey, TValue] = {}
        for key, value in other.items():
            if key in self._items:
                updated_items[key] = value
            else:
                added_items[key] = value
        self._items.update(other)

        # Notify for added items
        if added_items:
            for key, value in added_items.items():
                self._notify_add(key, value)

        # Notify for updated items
        if updated_items:
            for key, value in updated_items.items():
                self._notify_update(key, value)

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

        This method does not modify the dictionary or trigger any callbacks.

        Returns:
            A list of keys.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            keys = user_data.keys()  # Returns: ["name", "email"]
            ```
        """
        return list(self._items.keys())

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

        This method does not modify the dictionary or trigger any callbacks.

        Returns:
            A list of values.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            values = user_data.values()  # Returns: ["Alice", "alice@example.com"]
            ```
        """
        return list(self._items.values())

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

        This method does not modify the dictionary or trigger any callbacks.

        Returns:
            A list of (key, value) pairs.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            items = user_data.items()  # Returns: [("name", "Alice"), ("email", "alice@example.com")]
            ```
        """
        return list(self._items.items())

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

        This method returns a regular Python dictionary, not an ObservableDict.
        It does not modify the dictionary or trigger any callbacks.

        Returns:
            A shallow copy of the dictionary as a regular Python dictionary.

        Examples:
            ```python
            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            copy = user_data.copy()  # Returns: {"name": "Alice", "email": "alice@example.com"} as a regular dict
            ```
        """
        return self._items.copy()

    @override
    def on_change(self, callback: Callable[[ObservableDictChange[TKey, TValue]], None]) -> None:
        """
        Add a callback to be called when the dictionary changes.

        The callback will be called with an ObservableDictChange object that contains
        information about the type of change (add, remove, update, clear) and the affected items.

        Args:
            callback: A function that takes an ObservableDictChange object.

        Examples:
            ```python
            def on_dict_change(change):
                print(f"Change type: {change.type}")
                if change.type == ObservableCollectionChangeType.ADD:
                    print(f"Added key: {change.key}, value: {change.value}")
                elif change.type == ObservableCollectionChangeType.UPDATE:
                    print(f"Updated key: {change.key}, new value: {change.value}")

            user_data = ObservableDict[str, str]({"name": "Alice"})
            user_data.on_change(on_dict_change)
            user_data["email"] = "alice@example.com"  # Triggers the callback
            ```
        """
        self._change_callbacks.append(callback)

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

        This is a more specific alternative to on_change that only triggers
        for add operations and provides a simpler callback signature.

        Args:
            callback: A function that takes a key and value.

        Examples:
            ```python
            def on_item_added(key, value):
                print(f"Added {key}: {value}")

            user_data = ObservableDict[str, str]({"name": "Alice"})
            user_data.on_add(on_item_added)
            user_data["email"] = "alice@example.com"  # Triggers the callback with ("email", "alice@example.com")
            ```
        """
        self._add_callbacks.append(callback)

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

        This is a more specific alternative to on_change that only triggers
        for remove operations and provides a simpler callback signature.

        Args:
            callback: A function that takes a key and value.

        Examples:
            ```python
            def on_item_removed(key, value):
                print(f"Removed {key}: {value}")

            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            user_data.on_remove(on_item_removed)
            del user_data["email"]  # Triggers the callback with ("email", "alice@example.com")
            ```
        """
        self._remove_callbacks.append(callback)

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

        This is a more specific alternative to on_change that only triggers
        for update operations and provides a simpler callback signature.

        Args:
            callback: A function that takes a key and new value.

        Examples:
            ```python
            def on_item_updated(key, value):
                print(f"Updated {key} to {value}")

            user_data = ObservableDict[str, str]({"name": "Alice"})
            user_data.on_update(on_item_updated)
            user_data["name"] = "Alicia"  # Triggers the callback with ("name", "Alicia")
            ```
        """
        self._update_callbacks.append(callback)

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

        This is a more specific alternative to on_change that only triggers
        for clear operations and provides a simpler callback signature.

        Args:
            callback: A function that takes a dict of cleared items.

        Examples:
            ```python
            def on_dict_cleared(items):
                print(f"Cleared {len(items)} items: {items}")

            user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
            user_data.on_clear(on_dict_cleared)
            user_data.clear()  # Triggers the callback with {"name": "Alice", "email": "alice@example.com"}
            ```
        """
        self._clear_callbacks.append(callback)

    def _notify_add(self, key: TKey, value: TValue) -> None:
        """
        Notify all callbacks of an item being added.

        This internal method is called by methods that add items to the dictionary.
        It notifies both specific add callbacks and general change callbacks.

        Args:
            key: The key that was added.
            value: The value that was added.
        """
        # Call specific callbacks
        for callback in self._add_callbacks:
            callback(key, value)

        # Create a dictionary with the single item for the items field
        items_dict = {key: value}

        # Call general change callbacks
        change = ObservableDictChange(
            type=ObservableCollectionChangeType.ADD,
            key=key,
            value=value,
            items=items_dict,
        )
        for callback in self._change_callbacks:
            callback(change)

    def _notify_remove(self, key: TKey, value: TValue) -> None:
        """
        Notify all callbacks of an item being removed.

        This internal method is called by methods that remove items from the dictionary.
        It notifies both specific remove callbacks and general change callbacks.

        Args:
            key: The key that was removed.
            value: The value that was removed.
        """
        print(f"DEBUG: ObservableDict._notify_remove called with key={key}, value={value}")

        # Call specific callbacks
        for callback in self._remove_callbacks:
            callback(key, value)

        # Create a dictionary with the single item for the items field
        items_dict = {key: value}

        # Call general change callbacks
        change = ObservableDictChange(
            type=ObservableCollectionChangeType.REMOVE,
            key=key,
            value=value,
            items=items_dict,
        )
        print(f"DEBUG: ObservableDict._notify_remove - Created change object: type={change.type}, key={change.key}, value={change.value}")
        for callback in self._change_callbacks:
            callback(change)
        print("DEBUG: ObservableDict._notify_remove - Completed")

    def _notify_update(self, key: TKey, value: TValue) -> None:
        """
        Notify all callbacks of an item being updated.

        This internal method is called by methods that update items in the dictionary.
        It notifies both specific update callbacks and general change callbacks.

        Args:
            key: The key that was updated.
            value: The new value.
        """
        # Call specific callbacks
        for callback in self._update_callbacks:
            callback(key, value)

        # Create a dictionary with the single item for the items field
        items_dict = {key: value}

        # Call general change callbacks
        change = ObservableDictChange(
            type=ObservableCollectionChangeType.UPDATE,
            key=key,
            value=value,
            items=items_dict,
        )
        for callback in self._change_callbacks:
            callback(change)

    def _notify_clear(self, items: dict[TKey, TValue]) -> None:
        """
        Notify all callbacks of the dictionary being cleared.

        This internal method is called by the clear method.
        It notifies both specific clear callbacks and general change callbacks.

        Args:
            items: The items that were cleared.
        """
        # Call specific callbacks
        for callback in self._clear_callbacks:
            callback(items)

        # Call general change callbacks
        change = ObservableDictChange(type=ObservableCollectionChangeType.CLEAR, items=items)
        for callback in self._change_callbacks:
            callback(change)

__contains__(key)

Check if a key is in the dictionary.

Parameters:

Name Type Description Default
key TKey

The key to check for.

required

Returns:

Type Description
bool

True if the key is in the dictionary, False otherwise.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
if "name" in user_data:
    print("Name is present")
Source code in observant\observable_dict.py
@override
def __contains__(self, key: TKey) -> bool:
    """
    Check if a key is in the dictionary.

    Args:
        key: The key to check for.

    Returns:
        True if the key is in the dictionary, False otherwise.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        if "name" in user_data:
            print("Name is present")
        ```
    """
    return key in self._items

__delitem__(key)

Delete an item from the dictionary.

This method notifies callbacks about the removed item.

Parameters:

Name Type Description Default
key TKey

The key to delete.

required

Raises:

Type Description
KeyError

If the key is not found.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
del user_data["email"]  # Triggers remove callbacks
Source code in observant\observable_dict.py
@override
def __delitem__(self, key: TKey) -> None:
    """
    Delete an item from the dictionary.

    This method notifies callbacks about the removed item.

    Args:
        key: The key to delete.

    Raises:
        KeyError: If the key is not found.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        del user_data["email"]  # Triggers remove callbacks
        ```
    """
    value = self._items[key]
    del self._items[key]
    self._notify_remove(key, value)

__getitem__(key)

Get an item from the dictionary.

Parameters:

Name Type Description Default
key TKey

The key to look up.

required

Returns:

Type Description
TValue

The value for the key.

Raises:

Type Description
KeyError

If the key is not found.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
name = user_data["name"]  # Returns: "Alice"
Source code in observant\observable_dict.py
@override
def __getitem__(self, key: TKey) -> TValue:
    """
    Get an item from the dictionary.

    Args:
        key: The key to look up.

    Returns:
        The value for the key.

    Raises:
        KeyError: If the key is not found.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        name = user_data["name"]  # Returns: "Alice"
        ```
    """
    return self._items[key]

__init__(items=None, *, copy=False)

Initialize with optional external dict reference.

Parameters:

Name Type Description Default
items dict[TKey, TValue] | None

Optional external dict to observe. If None, creates a new dict.

None
copy bool

If True, creates a copy of the provided dict instead of using it directly. This is useful when you want to avoid modifying the original dict.

False

Examples:

# Create an empty observable dictionary
empty_dict = ObservableDict[str, int]()

# Create with initial items
user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})

# Create with a copy of initial items
original = {"red": "#FF0000", "green": "#00FF00"}
colors = ObservableDict[str, str](original, copy=True)
colors["blue"] = "#0000FF"  # original dict is not modified
Source code in observant\observable_dict.py
def __init__(self, items: dict[TKey, TValue] | None = None, *, copy: bool = False) -> None:
    """
    Initialize with optional external dict reference.

    Args:
        items: Optional external dict to observe. If None, creates a new dict.
        copy: If True, creates a copy of the provided dict instead of using it directly.
            This is useful when you want to avoid modifying the original dict.

    Examples:
        ```python
        # Create an empty observable dictionary
        empty_dict = ObservableDict[str, int]()

        # Create with initial items
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})

        # Create with a copy of initial items
        original = {"red": "#FF0000", "green": "#00FF00"}
        colors = ObservableDict[str, str](original, copy=True)
        colors["blue"] = "#0000FF"  # original dict is not modified
        ```
    """
    if copy:
        self._items: dict[TKey, TValue] = dict(items) if items is not None else {}
    else:
        self._items = items if items is not None else {}
    self._change_callbacks: list[Callable[[ObservableDictChange[TKey, TValue]], None]] = []
    self._add_callbacks: list[Callable[[TKey, TValue], None]] = []
    self._remove_callbacks: list[Callable[[TKey, TValue], None]] = []
    self._update_callbacks: list[Callable[[TKey, TValue], None]] = []
    self._clear_callbacks: list[Callable[[dict[TKey, TValue]], None]] = []

__iter__()

Return an iterator over the keys in the dictionary.

Returns:

Type Description
Iterator[TKey]

An iterator over the keys in the dictionary.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
for key in user_data:
    print(key)  # Prints: "name", "email"
Source code in observant\observable_dict.py
@override
def __iter__(self) -> Iterator[TKey]:
    """
    Return an iterator over the keys in the dictionary.

    Returns:
        An iterator over the keys in the dictionary.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        for key in user_data:
            print(key)  # Prints: "name", "email"
        ```
    """
    return iter(self._items)

__len__()

Return the number of items in the dictionary.

Returns:

Type Description
int

The number of items in the dictionary.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
length = len(user_data)  # Returns: 2
Source code in observant\observable_dict.py
@override
def __len__(self) -> int:
    """
    Return the number of items in the dictionary.

    Returns:
        The number of items in the dictionary.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        length = len(user_data)  # Returns: 2
        ```
    """
    return len(self._items)

__setitem__(key, value)

Set an item in the dictionary.

This method notifies callbacks about the added or updated item. If the key already exists, an update notification is sent. If the key is new, an add notification is sent.

Parameters:

Name Type Description Default
key TKey

The key to set.

required
value TValue

The value to set.

required

Examples:

user_data = ObservableDict[str, str]({"name": "Alice"})
user_data["email"] = "alice@example.com"  # Triggers add callbacks
user_data["name"] = "Alicia"  # Triggers update callbacks
Source code in observant\observable_dict.py
@override
def __setitem__(self, key: TKey, value: TValue) -> None:
    """
    Set an item in the dictionary.

    This method notifies callbacks about the added or updated item.
    If the key already exists, an update notification is sent.
    If the key is new, an add notification is sent.

    Args:
        key: The key to set.
        value: The value to set.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice"})
        user_data["email"] = "alice@example.com"  # Triggers add callbacks
        user_data["name"] = "Alicia"  # Triggers update callbacks
        ```
    """
    if key in self._items:
        self._items[key] = value
        self._notify_update(key, value)
    else:
        self._items[key] = value
        self._notify_add(key, value)

clear()

Remove all items from the dictionary.

This method notifies callbacks about the cleared items. If the dictionary is already empty, no notifications are sent.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
user_data.clear()  # Dictionary becomes {}
Source code in observant\observable_dict.py
@override
def clear(self) -> None:
    """
    Remove all items from the dictionary.

    This method notifies callbacks about the cleared items.
    If the dictionary is already empty, no notifications are sent.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        user_data.clear()  # Dictionary becomes {}
        ```
    """
    if not self._items:
        return
    items = self._items.copy()
    self._items.clear()
    self._notify_clear(items)

copy()

Return a shallow copy of the dictionary.

This method returns a regular Python dictionary, not an ObservableDict. It does not modify the dictionary or trigger any callbacks.

Returns:

Type Description
dict[TKey, TValue]

A shallow copy of the dictionary as a regular Python dictionary.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
copy = user_data.copy()  # Returns: {"name": "Alice", "email": "alice@example.com"} as a regular dict
Source code in observant\observable_dict.py
@override
def copy(self) -> dict[TKey, TValue]:
    """
    Return a shallow copy of the dictionary.

    This method returns a regular Python dictionary, not an ObservableDict.
    It does not modify the dictionary or trigger any callbacks.

    Returns:
        A shallow copy of the dictionary as a regular Python dictionary.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        copy = user_data.copy()  # Returns: {"name": "Alice", "email": "alice@example.com"} as a regular dict
        ```
    """
    return self._items.copy()

get(key, default=None)

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

This method does not modify the dictionary or trigger any callbacks.

Parameters:

Name Type Description Default
key TKey

The key to look up.

required
default TValue | None

The default value to return if the key is not found.

None

Returns:

Type Description
TValue | None

The value for the key, or the default value.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice"})
email = user_data.get("email", "No email")  # Returns: "No email"
name = user_data.get("name", "Unknown")  # Returns: "Alice"
Source code in observant\observable_dict.py
@override
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.

    This method does not modify the dictionary or trigger any callbacks.

    Args:
        key: The key to look up.
        default: The default value to return if the key is not found.

    Returns:
        The value for the key, or the default value.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice"})
        email = user_data.get("email", "No email")  # Returns: "No email"
        name = user_data.get("name", "Unknown")  # Returns: "Alice"
        ```
    """
    return self._items.get(key, default)

items()

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

This method does not modify the dictionary or trigger any callbacks.

Returns:

Type Description
list[tuple[TKey, TValue]]

A list of (key, value) pairs.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
items = user_data.items()  # Returns: [("name", "Alice"), ("email", "alice@example.com")]
Source code in observant\observable_dict.py
@override
def items(self) -> list[tuple[TKey, TValue]]:
    """
    Return a list of all (key, value) pairs in the dictionary.

    This method does not modify the dictionary or trigger any callbacks.

    Returns:
        A list of (key, value) pairs.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        items = user_data.items()  # Returns: [("name", "Alice"), ("email", "alice@example.com")]
        ```
    """
    return list(self._items.items())

keys()

Return a list of all keys in the dictionary.

This method does not modify the dictionary or trigger any callbacks.

Returns:

Type Description
list[TKey]

A list of keys.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
keys = user_data.keys()  # Returns: ["name", "email"]
Source code in observant\observable_dict.py
@override
def keys(self) -> list[TKey]:
    """
    Return a list of all keys in the dictionary.

    This method does not modify the dictionary or trigger any callbacks.

    Returns:
        A list of keys.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        keys = user_data.keys()  # Returns: ["name", "email"]
        ```
    """
    return list(self._items.keys())

on_add(callback)

Register for add events with key and value.

This is a more specific alternative to on_change that only triggers for add operations and provides a simpler callback signature.

Parameters:

Name Type Description Default
callback Callable[[TKey, TValue], None]

A function that takes a key and value.

required

Examples:

def on_item_added(key, value):
    print(f"Added {key}: {value}")

user_data = ObservableDict[str, str]({"name": "Alice"})
user_data.on_add(on_item_added)
user_data["email"] = "alice@example.com"  # Triggers the callback with ("email", "alice@example.com")
Source code in observant\observable_dict.py
@override
def on_add(self, callback: Callable[[TKey, TValue], None]) -> None:
    """
    Register for add events with key and value.

    This is a more specific alternative to on_change that only triggers
    for add operations and provides a simpler callback signature.

    Args:
        callback: A function that takes a key and value.

    Examples:
        ```python
        def on_item_added(key, value):
            print(f"Added {key}: {value}")

        user_data = ObservableDict[str, str]({"name": "Alice"})
        user_data.on_add(on_item_added)
        user_data["email"] = "alice@example.com"  # Triggers the callback with ("email", "alice@example.com")
        ```
    """
    self._add_callbacks.append(callback)

on_change(callback)

Add a callback to be called when the dictionary changes.

The callback will be called with an ObservableDictChange object that contains information about the type of change (add, remove, update, clear) and the affected items.

Parameters:

Name Type Description Default
callback Callable[[ObservableDictChange[TKey, TValue]], None]

A function that takes an ObservableDictChange object.

required

Examples:

def on_dict_change(change):
    print(f"Change type: {change.type}")
    if change.type == ObservableCollectionChangeType.ADD:
        print(f"Added key: {change.key}, value: {change.value}")
    elif change.type == ObservableCollectionChangeType.UPDATE:
        print(f"Updated key: {change.key}, new value: {change.value}")

user_data = ObservableDict[str, str]({"name": "Alice"})
user_data.on_change(on_dict_change)
user_data["email"] = "alice@example.com"  # Triggers the callback
Source code in observant\observable_dict.py
@override
def on_change(self, callback: Callable[[ObservableDictChange[TKey, TValue]], None]) -> None:
    """
    Add a callback to be called when the dictionary changes.

    The callback will be called with an ObservableDictChange object that contains
    information about the type of change (add, remove, update, clear) and the affected items.

    Args:
        callback: A function that takes an ObservableDictChange object.

    Examples:
        ```python
        def on_dict_change(change):
            print(f"Change type: {change.type}")
            if change.type == ObservableCollectionChangeType.ADD:
                print(f"Added key: {change.key}, value: {change.value}")
            elif change.type == ObservableCollectionChangeType.UPDATE:
                print(f"Updated key: {change.key}, new value: {change.value}")

        user_data = ObservableDict[str, str]({"name": "Alice"})
        user_data.on_change(on_dict_change)
        user_data["email"] = "alice@example.com"  # Triggers the callback
        ```
    """
    self._change_callbacks.append(callback)

on_clear(callback)

Register for clear events with the cleared items.

This is a more specific alternative to on_change that only triggers for clear operations and provides a simpler callback signature.

Parameters:

Name Type Description Default
callback Callable[[dict[TKey, TValue]], None]

A function that takes a dict of cleared items.

required

Examples:

def on_dict_cleared(items):
    print(f"Cleared {len(items)} items: {items}")

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
user_data.on_clear(on_dict_cleared)
user_data.clear()  # Triggers the callback with {"name": "Alice", "email": "alice@example.com"}
Source code in observant\observable_dict.py
@override
def on_clear(self, callback: Callable[[dict[TKey, TValue]], None]) -> None:
    """
    Register for clear events with the cleared items.

    This is a more specific alternative to on_change that only triggers
    for clear operations and provides a simpler callback signature.

    Args:
        callback: A function that takes a dict of cleared items.

    Examples:
        ```python
        def on_dict_cleared(items):
            print(f"Cleared {len(items)} items: {items}")

        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        user_data.on_clear(on_dict_cleared)
        user_data.clear()  # Triggers the callback with {"name": "Alice", "email": "alice@example.com"}
        ```
    """
    self._clear_callbacks.append(callback)

on_remove(callback)

Register for remove events with key and value.

This is a more specific alternative to on_change that only triggers for remove operations and provides a simpler callback signature.

Parameters:

Name Type Description Default
callback Callable[[TKey, TValue], None]

A function that takes a key and value.

required

Examples:

def on_item_removed(key, value):
    print(f"Removed {key}: {value}")

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
user_data.on_remove(on_item_removed)
del user_data["email"]  # Triggers the callback with ("email", "alice@example.com")
Source code in observant\observable_dict.py
@override
def on_remove(self, callback: Callable[[TKey, TValue], None]) -> None:
    """
    Register for remove events with key and value.

    This is a more specific alternative to on_change that only triggers
    for remove operations and provides a simpler callback signature.

    Args:
        callback: A function that takes a key and value.

    Examples:
        ```python
        def on_item_removed(key, value):
            print(f"Removed {key}: {value}")

        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        user_data.on_remove(on_item_removed)
        del user_data["email"]  # Triggers the callback with ("email", "alice@example.com")
        ```
    """
    self._remove_callbacks.append(callback)

on_update(callback)

Register for update events with key and new value.

This is a more specific alternative to on_change that only triggers for update operations and provides a simpler callback signature.

Parameters:

Name Type Description Default
callback Callable[[TKey, TValue], None]

A function that takes a key and new value.

required

Examples:

def on_item_updated(key, value):
    print(f"Updated {key} to {value}")

user_data = ObservableDict[str, str]({"name": "Alice"})
user_data.on_update(on_item_updated)
user_data["name"] = "Alicia"  # Triggers the callback with ("name", "Alicia")
Source code in observant\observable_dict.py
@override
def on_update(self, callback: Callable[[TKey, TValue], None]) -> None:
    """
    Register for update events with key and new value.

    This is a more specific alternative to on_change that only triggers
    for update operations and provides a simpler callback signature.

    Args:
        callback: A function that takes a key and new value.

    Examples:
        ```python
        def on_item_updated(key, value):
            print(f"Updated {key} to {value}")

        user_data = ObservableDict[str, str]({"name": "Alice"})
        user_data.on_update(on_item_updated)
        user_data["name"] = "Alicia"  # Triggers the callback with ("name", "Alicia")
        ```
    """
    self._update_callbacks.append(callback)

pop(key, default=None)

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

If the key exists, this method removes it and triggers remove callbacks. If the key does not exist and a default is provided, no callbacks are triggered.

Parameters:

Name Type Description Default
key TKey

The key to look up.

required
default TValue | None

The default value to return if the key is not found.

None

Returns:

Type Description
TValue | None

The value for the key, or the default value.

Raises:

Type Description
KeyError

If the key is not found and no default value is provided.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
email = user_data.pop("email")  # Returns: "alice@example.com" and removes it
phone = user_data.pop("phone", "No phone")  # Returns: "No phone" without modifying the dict
Source code in observant\observable_dict.py
@override
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.

    If the key exists, this method removes it and triggers remove callbacks.
    If the key does not exist and a default is provided, no callbacks are triggered.

    Args:
        key: The key to look up.
        default: The default value to return if the key is not found.

    Returns:
        The value for the key, or the default value.

    Raises:
        KeyError: If the key is not found and no default value is provided.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        email = user_data.pop("email")  # Returns: "alice@example.com" and removes it
        phone = user_data.pop("phone", "No phone")  # Returns: "No phone" without modifying the dict
        ```
    """
    if key in self._items:
        value = self._items.pop(key)
        self._notify_remove(key, value)
        return value
    if default is not None:
        return default
    raise KeyError(key)

popitem()

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

This method removes an arbitrary (key, value) pair and triggers remove callbacks. In Python 3.7+, the pairs are returned in LIFO order (last inserted, first returned).

Returns:

Type Description
tuple[TKey, TValue]

A (key, value) pair.

Raises:

Type Description
KeyError

If the dictionary is empty.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
key, value = user_data.popitem()  # Might return: ("email", "alice@example.com")
Source code in observant\observable_dict.py
@override
def popitem(self) -> tuple[TKey, TValue]:
    """
    Remove and return a (key, value) pair from the dictionary.

    This method removes an arbitrary (key, value) pair and triggers remove callbacks.
    In Python 3.7+, the pairs are returned in LIFO order (last inserted, first returned).

    Returns:
        A (key, value) pair.

    Raises:
        KeyError: If the dictionary is empty.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        key, value = user_data.popitem()  # Might return: ("email", "alice@example.com")
        ```
    """
    key, value = self._items.popitem()
    self._notify_remove(key, value)
    return key, value

setdefault(key, default=None)

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

If the key does not exist, this method adds it with the default value and triggers add callbacks.

Parameters:

Name Type Description Default
key TKey

The key to look up.

required
default TValue | None

The default value to set and return if the key is not found.

None

Returns:

Type Description
TValue | None

The value for the key, or the default value.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice"})
email = user_data.setdefault("email", "alice@example.com")  # Returns: "alice@example.com" and adds it
name = user_data.setdefault("name", "Unknown")  # Returns: "Alice" without changing it
Source code in observant\observable_dict.py
@override
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.

    If the key does not exist, this method adds it with the default value and
    triggers add callbacks.

    Args:
        key: The key to look up.
        default: The default value to set and return if the key is not found.

    Returns:
        The value for the key, or the default value.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice"})
        email = user_data.setdefault("email", "alice@example.com")  # Returns: "alice@example.com" and adds it
        name = user_data.setdefault("name", "Unknown")  # Returns: "Alice" without changing it
        ```
    """
    if key not in self._items:
        self._items[key] = cast(TValue, default)  # Cast to V since we know it's a value
        self._notify_add(key, cast(TValue, default))
        return default
    return self._items[key]

update(other)

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

This method notifies callbacks about added and updated items. For each key in the other dictionary: - If the key already exists, an update notification is sent. - If the key is new, an add notification is sent. If the other dictionary is empty, no notifications are sent.

Parameters:

Name Type Description Default
other dict[TKey, TValue]

Another dictionary to update from.

required

Examples:

user_data = ObservableDict[str, str]({"name": "Alice"})
user_data.update({"email": "alice@example.com", "name": "Alicia"})
# Dictionary becomes {"name": "Alicia", "email": "alice@example.com"}
Source code in observant\observable_dict.py
@override
def update(self, other: dict[TKey, TValue]) -> None:
    """
    Update the dictionary with the key/value pairs from another dictionary.

    This method notifies callbacks about added and updated items.
    For each key in the other dictionary:
    - If the key already exists, an update notification is sent.
    - If the key is new, an add notification is sent.
    If the other dictionary is empty, no notifications are sent.

    Args:
        other: Another dictionary to update from.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice"})
        user_data.update({"email": "alice@example.com", "name": "Alicia"})
        # Dictionary becomes {"name": "Alicia", "email": "alice@example.com"}
        ```
    """
    if not other:
        return
    added_items: dict[TKey, TValue] = {}
    updated_items: dict[TKey, TValue] = {}
    for key, value in other.items():
        if key in self._items:
            updated_items[key] = value
        else:
            added_items[key] = value
    self._items.update(other)

    # Notify for added items
    if added_items:
        for key, value in added_items.items():
            self._notify_add(key, value)

    # Notify for updated items
    if updated_items:
        for key, value in updated_items.items():
            self._notify_update(key, value)

values()

Return a list of all values in the dictionary.

This method does not modify the dictionary or trigger any callbacks.

Returns:

Type Description
list[TValue]

A list of values.

Examples:

user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
values = user_data.values()  # Returns: ["Alice", "alice@example.com"]
Source code in observant\observable_dict.py
@override
def values(self) -> list[TValue]:
    """
    Return a list of all values in the dictionary.

    This method does not modify the dictionary or trigger any callbacks.

    Returns:
        A list of values.

    Examples:
        ```python
        user_data = ObservableDict[str, str]({"name": "Alice", "email": "alice@example.com"})
        values = user_data.values()  # Returns: ["Alice", "alice@example.com"]
        ```
    """
    return list(self._items.values())