import weakref
from contextlib import ExitStack, contextmanager
from itertools import chain
from weakref import WeakKeyDictionary
from .alias import CallbackPropertyAlias
from .callback_container import CallbackContainer
__all__ = [
"CallbackProperty",
"callback_property",
"add_callback",
"remove_callback",
"delay_callback",
"ignore_callback",
"HasCallbackProperties",
"keep_in_sync",
]
[docs]
class CallbackProperty:
"""
A property that callback functions can be added to.
When a callback property changes value, each callback function
is called with information about the state change. Otherwise,
callback properties behave just like normal instance variables.
CallbackProperties must be defined at the class level. Use
the helper function :func:`~echo.add_callback` to attach a callback to
a specific instance of a class with CallbackProperties
Parameters
----------
default
The initial value for the property
docstring : str
The docstring for the property
getter, setter : func
Custom getter and setter functions (advanced)
"""
def __init__(self, default=None, docstring=None, getter=None, setter=None):
"""
:param default: The initial value for the property
"""
self._default = default
self._validators = WeakKeyDictionary()
self._2arg_validators = WeakKeyDictionary()
self._callbacks = WeakKeyDictionary()
self._2arg_callbacks = WeakKeyDictionary()
self._disabled = WeakKeyDictionary()
self._values = WeakKeyDictionary()
if getter is None:
getter = self._default_getter
if setter is None:
setter = self._default_setter
self._getter = getter
self._setter = setter
if docstring is not None:
self.__doc__ = docstring
def _default_getter(self, instance, owner=None):
return self._values.get(instance, self._default)
def _default_setter(self, instance, value):
self._values.__setitem__(instance, value)
def __get__(self, instance, owner=None):
if instance is None:
return self
return self._getter(instance)
def __set__(self, instance, value):
try:
old = self.__get__(instance)
except AttributeError: # pragma: no cover
old = None
value = self._validate(instance, old, value)
self._setter(instance, value)
new = self.__get__(instance)
if old != new:
self.notify(instance, old, new)
[docs]
def setter(self, func):
"""
Method to use as a decorator, to mimic @property.setter
"""
self._setter = func
return self
def _get_full_info(self, instance):
# Some callback subclasses may contain additional info in addition
# to the main value, and we need to use this full information when
# comparing old and new 'values', so this method is used in that
# case. The result should be a tuple where the first item is the
# actual primary value of the property and the second item is any
# additional data to use in the comparison.
# Note that we need to make sure we convert any list here to a tuple
# to make sure the value is immutable, otherwise comparisons of
# old != new will not show any difference (since the list can still)
# be modified in-place
value = self.__get__(instance)
if isinstance(value, list):
value = tuple(value)
return value, None
[docs]
def notify(self, instance, old, new):
"""
Call all callback functions with the current value
Each callback will either be called using
callback(new) or callback(old, new) depending
on whether ``echo_old`` was set to `True` when calling
:func:`~echo.add_callback`
Parameters
----------
instance
The instance to consider
old
The old value of the property
new
The new value of the property
"""
if not self.enabled(instance):
return
iterators = []
if container := self._callbacks.get(instance, None):
iterators.append((cb, priority, 1) for cb, priority in container.iterator(priority=True, sort=False))
if container := self._2arg_callbacks.get(instance, None):
iterators.append((cb, priority, 2) for cb, priority in container.iterator(priority=True, sort=False))
for callback, _priority, args in sorted(chain(*iterators), key=lambda x: x[1], reverse=True):
if args == 1:
callback(new)
else:
callback(old, new)
def _validate(self, instance, old, new):
"""
Call all validators.
Each validator will either be called using
validator(new) or validator(old, new) depending
on whether ``echo_old`` was set to `True` when calling
:func:`~echo.add_callback`
Parameters
----------
instance
The instance to consider
old
The old value of the property
new
The new value of the property
"""
# Note: validators can't be delayed so we don't check for
# enabled/disabled as in notify()
for cback in self._validators.get(instance, []):
new = cback(new)
for cback in self._2arg_validators.get(instance, []):
new = cback(old, new)
return new
[docs]
def disable(self, instance):
"""
Disable callbacks for a specific instance
"""
self._disabled[instance] = True
[docs]
def enable(self, instance):
"""
Enable previously-disabled callbacks for a specific instance
"""
self._disabled[instance] = False
[docs]
def enabled(self, instance):
return not self._disabled.get(instance, False)
[docs]
def add_callback(self, instance, func, echo_old=False, priority=0, validator=False):
"""
Add a callback to a specific instance that manages this property
Parameters
----------
instance
The instance to add the callback to
func : func
The callback function to add
echo_old : bool, optional
If `True`, the callback function will be invoked with both the old
and new values of the property, as ``func(old, new)``. If `False`
(the default), will be invoked as ``func(new)``
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 *before* the property is set. The
validator can return a modified value (for example it can be used
to change the types of values or change properties in-place) or it
can also raise an exception.
"""
if validator:
if echo_old:
self._2arg_validators.setdefault(instance, CallbackContainer()).append(func, priority=priority)
else:
self._validators.setdefault(instance, CallbackContainer()).append(func, priority=priority)
else:
if echo_old:
self._2arg_callbacks.setdefault(instance, CallbackContainer()).append(func, priority=priority)
else:
self._callbacks.setdefault(instance, CallbackContainer()).append(func, priority=priority)
@property
def _all_callbacks(self):
return [self._validators, self._2arg_validators, self._callbacks, self._2arg_callbacks]
[docs]
def remove_callback(self, instance, func):
"""
Remove a previously-added callback
Parameters
----------
instance
The instance to detach the callback from
func : func
The callback function to remove
"""
for cb in self._all_callbacks:
if instance not in cb:
continue
if func in cb[instance]:
cb[instance].remove(func)
return
else:
raise ValueError(f"Callback function not found: {func}")
[docs]
def clear_callbacks(self, instance):
"""
Remove all callbacks on this property.
"""
for cb in self._all_callbacks:
if instance in cb:
cb[instance].clear()
if instance in self._disabled:
self._disabled.pop(instance)
[docs]
class HasCallbackProperties:
"""
A class that adds functionality to subclasses that use callback properties.
"""
def __init__(self):
from .containers import DictCallbackProperty, ListCallbackProperty
self._global_callbacks = CallbackContainer()
self._ignored_properties = set()
self._delayed_properties = {}
self._delay_global_calls = {}
self._callback_wrappers = {}
self._notify_context_managers = []
for prop_name, prop in self.iter_callback_properties():
if isinstance(prop, ListCallbackProperty | DictCallbackProperty):
prop.add_callback(self, self._notify_global_listordict)
def _ignore_global_callbacks(self, properties):
# This is to allow ignore_callbacks to work for global callbacks
self._ignored_properties.update(properties)
def _unignore_global_callbacks(self, properties):
# Once this is called, we simply remove properties from _ignored_properties
# and don't call the callbacks. This is used by ignore_callback
self._ignored_properties -= set(properties)
def _delay_global_callbacks(self, properties):
# This is to allow delay_callback to still have an effect in delaying
# global callbacks. We set _delayed_properties to a dictionary of the
# values at the point at which the callbacks are delayed.
self._delayed_properties.update(properties)
def _process_delayed_global_callbacks(self, properties):
# Once this is called, the global callbacks are called once each with
# a dictionary of the current values of properties that have been
# resumed.
kwargs = {}
for prop, new_value in properties.items():
old_value = self._delayed_properties.pop(prop)
if old_value != new_value:
kwargs[prop] = new_value[0]
self._notify_global(**kwargs)
def _notify_global_listordict(self, *args):
from .containers import DictCallbackProperty, ListCallbackProperty
properties = {}
for prop_name, prop in self.iter_callback_properties():
if isinstance(prop, ListCallbackProperty | DictCallbackProperty):
callback_listordict = getattr(self, prop_name)
if callback_listordict is args[0]:
properties[prop_name] = callback_listordict
break
self._notify_global(**properties)
def _get_aliases_for(self, name):
"""
Return a list of alias names that point to the given property name.
"""
return [
attr_name
for attr_name in dir(type(self))
if self.is_alias(attr_name) and getattr(type(self), attr_name)._target == name
]
def _notify_global(self, **kwargs):
# Add aliases for any properties being notified (for backward compatibility)
kwargs.update(
{alias: value for prop_name, value in kwargs.items() for alias in self._get_aliases_for(prop_name)}
)
for prop in set(self._delayed_properties) | set(self._ignored_properties):
if prop in kwargs:
kwargs.pop(prop)
if len(kwargs) > 0:
for callback in self._global_callbacks:
callback(**kwargs)
def __setattr__(self, attribute, value):
# Trigger global callbacks for callback properties, but not aliases
# (aliases will trigger via the target property they redirect to)
is_callback = self.is_callback_property(attribute) and not self.is_alias(attribute)
if is_callback:
previous_value = getattr(self, attribute, None)
super().__setattr__(attribute, value)
if is_callback and value != previous_value:
self._notify_global(**{attribute: value})
[docs]
def add_callback(self, name, callback, echo_old=False, priority=0, validator=False):
"""
Add a callback that gets triggered when a callback property of the
class changes.
Parameters
----------
name : str
The instance to add the callback to.
callback : func
The callback function to add
echo_old : bool, optional
If `True`, the callback function will be invoked with both the old
and new values of the property, as ``callback(old, new)``. If `False`
(the default), will be invoked as ``callback(new)``
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 *before* the property is set. The
validator can return a modified value (for example it can be used
to change the types of values or change properties in-place) or it
can also raise an exception."""
if self.is_callback_property(name):
prop = getattr(type(self), name)
if self.is_alias(name):
prop._warn()
prop = prop._target_property
prop.add_callback(self, callback, echo_old=echo_old, priority=priority, validator=validator)
else:
raise TypeError(f"attribute '{name}' is not a callback property")
[docs]
def remove_callback(self, name, callback):
"""
Remove a previously-added callback
Parameters
----------
name : str
The instance to remove the callback from.
func : func
The callback function to remove
"""
if self.is_callback_property(name):
prop = getattr(type(self), name)
if self.is_alias(name):
prop._warn()
prop = prop._target_property
try:
prop.remove_callback(self, callback)
except ValueError: # pragma: nocover
pass # Be forgiving if callback was already removed before
else:
raise TypeError(f"attribute '{name}' is not a callback property")
[docs]
def add_global_callback(self, callback):
"""
Add a global callback function, which is a callback that gets triggered
when any callback properties on the class change.
Parameters
----------
callback : func
The callback function to add
"""
self._global_callbacks.append(callback)
[docs]
def remove_global_callback(self, callback):
"""
Remove a global callback function.
Parameters
----------
callback : func
The callback function to remove
"""
self._global_callbacks.remove(callback)
[docs]
def is_callback_property(self, name):
"""
Whether a property (identified by name) is a callback property.
Returns True for both CallbackProperty and CallbackPropertyAlias.
Parameters
----------
name : str
The name of the property to check
"""
prop = getattr(type(self), name, None)
return isinstance(prop, CallbackProperty | CallbackPropertyAlias)
[docs]
def is_alias(self, name):
"""
Whether a property (identified by name) is a callback property alias.
Parameters
----------
name : str
The name of the property to check
"""
prop = getattr(type(self), name, None)
return isinstance(prop, CallbackPropertyAlias)
[docs]
def iter_callback_properties(self):
"""
Iterator to loop over all callback properties.
Note: This does not include CallbackPropertyAlias instances, only
actual CallbackProperty instances.
"""
for name in dir(self):
prop = getattr(type(self), name, None)
if isinstance(prop, CallbackProperty):
yield name, prop
[docs]
def callback_properties(self):
"""
Return a list of all callback property names.
Note: This does not include CallbackPropertyAlias instances, only
actual CallbackProperty instances.
"""
return [name for name, _ in self.iter_callback_properties()]
[docs]
def clear_callbacks(self):
"""
Remove all global and property-specific callbacks.
"""
self._global_callbacks.clear()
for name, prop in self.iter_callback_properties():
prop.clear_callbacks(self)
def _resolve_callback_property(instance, prop):
"""
Resolve a property name to its CallbackProperty, handling aliases.
If the property is a CallbackPropertyAlias, follows it to the target
property and emits a deprecation warning if applicable.
Returns
-------
tuple
(resolved_name, callback_property) - the resolved property name and
the actual CallbackProperty object
"""
p = getattr(type(instance), prop)
if isinstance(p, CallbackPropertyAlias):
p._warn()
return p._target, p._target_property
return prop, p
[docs]
def add_callback(instance, prop, callback, echo_old=False, priority=0, validator=False):
"""
Attach a callback function to a property in an instance
Parameters
----------
instance
The instance to add the callback to
prop : str
Name of callback property in `instance`
callback : func
The callback function to add
echo_old : bool, optional
If `True`, the callback function will be invoked with both the old
and new values of the property, as ``func(old, new)``. If `False`
(the default), will be invoked as ``func(new)``
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 *before* the property is set. The
validator can return a modified value (for example it can be used
to change the types of values or change properties in-place) or it
can also raise an exception.
Examples
--------
::
class Foo:
bar = CallbackProperty(0)
def callback(value):
pass
f = Foo()
add_callback(f, 'bar', callback)
"""
prop, p = _resolve_callback_property(instance, prop)
if not isinstance(p, CallbackProperty):
raise TypeError(f"{prop} is not a CallbackProperty")
p.add_callback(instance, callback, echo_old=echo_old, priority=priority, validator=validator)
[docs]
def remove_callback(instance, prop, callback):
"""
Remove a callback function from a property in an instance
Parameters
----------
instance
The instance to detach the callback from
prop : str
Name of callback property in `instance`
callback : func
The callback function to remove
"""
prop, p = _resolve_callback_property(instance, prop)
if not isinstance(p, CallbackProperty):
raise TypeError(f"{prop} is not a CallbackProperty")
p.remove_callback(instance, callback)
[docs]
def callback_property(getter):
"""
A decorator to build a CallbackProperty.
This is used by wrapping a getter method, similar to the use of @property::
class Foo(object):
@callback_property
def x(self):
return self._x
@x.setter
def x(self, value):
self._x = value
In simple cases with no getter or setter logic, it's easier to create a
:class:`~echo.CallbackProperty` directly::
class Foo(object);
x = CallbackProperty(initial_value)
"""
cb = CallbackProperty(getter=getter)
cb.__doc__ = getter.__doc__
return cb
[docs]
class delay_callback:
"""
Delay any callback functions from one or more callback properties
This is a context manager. Within the context block, no callbacks
will be issued. Each callback will be called once on exit
Parameters
----------
instance
An instance object with callback properties
*props : str
One or more properties within instance to delay
Examples
--------
::
with delay_callback(foo, 'bar', 'baz'):
f.bar = 20
f.baz = 30
f.bar = 10
print('done') # callbacks triggered at this point, if needed
"""
# Class-level registry of properties and how many times the callbacks have
# been delayed. The idea is that when nesting calls to delay_callback, the
# delay count is increased, and every time __exit__ is called, the count is
# decreased, and once the count reaches zero, the callback is triggered.
delay_count = {}
old_values = {}
def __init__(self, instance, *props):
self.instance = instance
# Resolve any aliases to their target property names
self.props = tuple(_resolve_callback_property(instance, p)[0] for p in props)
def __enter__(self):
delay_props = {}
for prop in self.props:
p = getattr(type(self.instance), prop)
if not isinstance(p, CallbackProperty):
raise TypeError(f"{prop} is not a CallbackProperty")
if (self.instance, prop) not in self.delay_count:
self.delay_count[self.instance, prop] = 1
self.old_values[self.instance, prop] = p._get_full_info(self.instance)
delay_props[prop] = p._get_full_info(self.instance)
else:
self.delay_count[self.instance, prop] += 1
p.disable(self.instance)
if isinstance(self.instance, HasCallbackProperties):
self.instance._delay_global_callbacks(delay_props)
def __exit__(self, *args):
resume_props = {}
notifications = []
for prop in self.props:
p = getattr(type(self.instance), prop)
if not isinstance(p, CallbackProperty): # pragma: no cover
raise TypeError(f"{prop} is not a CallbackProperty")
if self.delay_count[self.instance, prop] > 1:
self.delay_count[self.instance, prop] -= 1
else:
self.delay_count.pop((self.instance, prop))
old = self.old_values.pop((self.instance, prop))
p.enable(self.instance)
new = p._get_full_info(self.instance)
if old != new:
notifications.append((p, (self.instance, old[0], new[0])))
resume_props[prop] = new
if isinstance(self.instance, HasCallbackProperties):
self.instance._process_delayed_global_callbacks(resume_props)
if cms := getattr(self.instance, "_notify_context_managers", None):
with ExitStack() as stack:
for cm_factory in cms:
stack.enter_context(cm_factory())
for p, args in notifications:
p.notify(*args)
else:
for p, args in notifications:
p.notify(*args)
[docs]
@contextmanager
def ignore_callback(instance, *props):
"""
Temporarily ignore any callbacks from one or more callback properties
This is a context manager. Within the context block, no callbacks will be
issued. In contrast with `delay_callback`, no callbacks will be
called on exiting the context manager
Parameters
----------
instance
An instance object with callback properties
*props : str
One or more properties within instance to ignore
Examples
--------
::
with ignore_callback(foo, 'bar', 'baz'):
f.bar = 20
f.baz = 30
f.bar = 10
print('done') # no callbacks called
"""
# Resolve any aliases to their target property names
props = tuple(_resolve_callback_property(instance, p)[0] for p in props)
for prop in props:
p = getattr(type(instance), prop)
if not isinstance(p, CallbackProperty):
raise TypeError(f"{prop} is not a CallbackProperty")
p.disable(instance)
if isinstance(instance, HasCallbackProperties):
instance._ignore_global_callbacks(props)
yield
for prop in props:
p = getattr(type(instance), prop)
assert isinstance(p, CallbackProperty)
p.enable(instance)
if isinstance(instance, HasCallbackProperties):
instance._unignore_global_callbacks(props)
[docs]
class keep_in_sync:
def __init__(self, instance1, prop1, instance2, prop2):
self.instance1 = weakref.ref(instance1, self.disable_syncing)
self.prop1 = prop1
self.instance2 = weakref.ref(instance2, self.disable_syncing)
self.prop2 = prop2
self._syncing = False
self.enabled = False
self.enable_syncing()
[docs]
def prop1_from_prop2(self, value):
if not self._syncing:
self._syncing = True
setattr(self.instance1(), self.prop1, getattr(self.instance2(), self.prop2))
self._syncing = False
[docs]
def prop2_from_prop1(self, value):
if not self._syncing:
self._syncing = True
setattr(self.instance2(), self.prop2, getattr(self.instance1(), self.prop1))
self._syncing = False
[docs]
def enable_syncing(self, *args):
if self.enabled:
return
add_callback(self.instance1(), self.prop1, self.prop2_from_prop1)
add_callback(self.instance2(), self.prop2, self.prop1_from_prop2)
self.enabled = True
[docs]
def disable_syncing(self, *args):
if not self.enabled:
return
if self.instance1() is not None:
remove_callback(self.instance1(), self.prop1, self.prop2_from_prop1)
if self.instance2() is not None:
remove_callback(self.instance2(), self.prop2, self.prop1_from_prop2)
self.enabled = False