import weakref
from . import CallbackProperty, HasCallbackProperties
from .callback_container import CallbackContainer
__all__ = ["CallbackList", "CallbackDict", "ListCallbackProperty", "DictCallbackProperty"]
class ContainerMixin:
def _setup_container(self):
self._callbacks = CallbackContainer()
self._item_validators = CallbackContainer()
def _prepare_add(self, value):
for validator in self._item_validators:
value = validator(value)
if isinstance(value, list):
value = CallbackList(self.notify_all, value)
elif isinstance(value, dict):
value = CallbackDict(self.notify_all, value)
if isinstance(value, HasCallbackProperties):
value.add_global_callback(self.notify_all)
elif isinstance(value, CallbackList | CallbackDict):
value.callback = self.notify_all
return value
def _cleanup_remove(self, value):
if isinstance(value, HasCallbackProperties):
value.remove_global_callback(self.notify_all)
elif isinstance(value, CallbackList | CallbackDict):
value.remove_callback(self.notify_all)
def add_callback(self, func, priority=0, validator=False):
"""
Add a callback to the container.
Note that validators are applied on a per item basis, whereas regular
callbacks are called with the whole list after modification.
Parameters
----------
func : func
The callback function to add
priority : int, optional
This can optionally be used to force a certain order of execution of
callbacks (larger values indicate a higher priority).
validator : bool, optional
Whether the callback is a validator, which is a special kind of
callback that gets called with the item being added to the
container *before* the container is modified. The validator can
return the value as-is, modify it, or emit warnings or an exception.
"""
if validator:
self._item_validators.append(func, priority=priority)
else:
self._callbacks.append(func, priority=priority)
def remove_callback(self, func):
"""
Remove a callback from the container.
"""
for cb in (self._callbacks, self._item_validators):
if func in cb:
cb.remove(func)
def notify_all(self, *args, **kwargs):
for callback in self._callbacks:
callback(*args, **kwargs)
[docs]
class CallbackList(list, ContainerMixin):
"""
A list that calls a callback function when it is modified.
The first argument should be the callback function (which takes no
arguments), and subsequent arguments are as for `list`.
"""
def __init__(self, callback, *args, **kwargs):
super().__init__(*args, **kwargs)
self._setup_container()
self.add_callback(callback)
for index, value in enumerate(self):
super().__setitem__(index, self._prepare_add(value))
def __repr__(self):
return f"<CallbackList with {len(self)} elements>"
[docs]
def append(self, value):
super().append(self._prepare_add(value))
self.notify_all()
[docs]
def extend(self, iterable):
iterable = [self._prepare_add(value) for value in iterable]
super().extend(iterable)
self.notify_all()
[docs]
def insert(self, index, value):
super().insert(index, self._prepare_add(value))
self.notify_all()
[docs]
def pop(self, index=-1):
result = super().pop(index)
self._cleanup_remove(result)
self.notify_all()
return result
[docs]
def remove(self, value):
super().remove(value)
self._cleanup_remove(value)
self.notify_all()
[docs]
def reverse(self):
super().reverse()
self.notify_all()
[docs]
def sort(self, key=None, reverse=False):
super().sort(key=key, reverse=reverse)
self.notify_all()
def __setitem__(self, slc, new_value):
old_values = self[slc]
if not isinstance(slc, slice):
old_values = [old_values]
for old_value in old_values:
self._cleanup_remove(old_value)
if isinstance(slc, slice):
new_value = [self._prepare_add(value) for value in new_value]
else:
new_value = self._prepare_add(new_value)
super().__setitem__(slc, new_value)
self.notify_all()
[docs]
def clear(self):
for item in self:
self._cleanup_remove(item)
super().clear()
self.notify_all()
[docs]
class CallbackDict(dict, ContainerMixin):
"""
A dictionary that calls a callback function when it is modified.
The first argument should be the callback function (which takes no
arguments), and subsequent arguments are passed to `dict`.
"""
def __init__(self, callback, *args, **kwargs):
super().__init__(*args, **kwargs)
self._setup_container()
self.add_callback(callback)
for key, value in self.items():
super().__setitem__(key, self._prepare_add(value))
[docs]
def clear(self):
for value in self.values():
self._cleanup_remove(value)
super().clear()
self.notify_all()
[docs]
def popitem(self):
result = super().popitem()
self._cleanup_remove(result)
self.notify_all()
return result
[docs]
def update(self, *args, **kwargs):
values = {}
values.update(*args, **kwargs)
for key, value in values.items():
values[key] = self._prepare_add(value)
super().update(values)
self.notify_all()
[docs]
def pop(self, *args, **kwargs):
result = super().pop(*args, **kwargs)
self._cleanup_remove(result)
self.notify_all()
return result
def __setitem__(self, key, value):
if key in self:
self._cleanup_remove(self[key])
super().__setitem__(key, self._prepare_add(value))
self.notify_all()
def __repr__(self):
return f"<CallbackDict with {len(self)} elements>"
class dynamic_callback:
function = None
def __call__(self, *args, **kwargs):
self.function(*args, **kwargs)
[docs]
class ListCallbackProperty(CallbackProperty):
"""
A list property that calls callbacks when its contents are modified
"""
def _default_getter(self, instance, owner=None):
if instance not in self._values:
self._default_setter(instance, self._default or [])
return super()._default_getter(instance, owner)
def _default_setter(self, instance, value):
if not isinstance(value, list):
raise TypeError("callback property should be a list")
dcb = dynamic_callback()
wrapped_list = CallbackList(dcb, value)
_instance = weakref.ref(instance)
def callback(*args, **kwargs):
if _instance() is not None:
self.notify(_instance(), wrapped_list, wrapped_list)
dcb.function = callback
super()._default_setter(instance, wrapped_list)
[docs]
class DictCallbackProperty(CallbackProperty):
"""
A dictionary property that calls callbacks when its contents are modified
"""
def _default_getter(self, instance, owner=None):
if instance not in self._values:
self._default_setter(instance, self._default or {})
return super()._default_getter(instance, owner)
def _default_setter(self, instance, value):
if not isinstance(value, dict):
raise TypeError("Callback property should be a dictionary.")
dcb = dynamic_callback()
wrapped_dict = CallbackDict(dcb, value)
_instance = weakref.ref(instance)
def callback(*args, **kwargs):
if _instance() is not None:
self.notify(_instance(), wrapped_dict, wrapped_dict)
dcb.function = callback
super()._default_setter(instance, wrapped_dict)