observers and dispatchers

dispatcher.py

from more_itertools import always_iterable
from sys import modules
from weakref import WeakKeyDictionary
from types import FunctionType

class WeakDefaultDict(WeakKeyDictionary):
    def __init__(self, default_factory=lambda: None, **kwargs):
        self._default_factory = default_factory
        super().__init__(**kwargs)

    def __getitem__(self, key):
        if key not in self:
            super().__setitem__(key, self._default_factory())
        return super().__getitem__(key)


class Property:
    def __init__(self, default=None):
        self.default = default
        self._instance_to_callbacks = WeakDefaultDict(list)

    def __set_name__(self, owner, name):
        self.name = name

    def __set__(self, instance, value):
        old = instance.__dict__[self.name]
        instance.__dict__[self.name] = value
        for callback in self._instance_to_callbacks[instance]:
            callback(instance, self.name, old, value)

    def __repr__(self):
        return f'{self.__class__.__name__}(default={self.default})'

    def bind(self, instance, callback):
        self._instance_to_callbacks[instance].append(callback)

    def bind_deco(self, instance):
        def deco(func):
            self.bind(instance, func)
            return func
        return deco


class MISSING:
    def __init__(self, name=''):
        self.name = name

    def __repr__(self):
        return 'MISSING'


class MISSINGDefaultDict(dict):
    def __missing__(self, key):
        if key.startswith('__'):
            raise KeyError(key)

        self[key] = MISSING(key)
        return self[key]


class DispatcherMeta(type):
    def __prepare__(*args, **kwargs):
        return MISSINGDefaultDict()

    def __new__(meta, name, bases, methods, **kwargs):
        module = modules[methods['__module__']]
        methods  = {name_: attr for name_, attr in methods.items()
                    if not isinstance(attr, MISSING) or not hasattr(module, name_)}

        for name_, funcs in methods.get('__annotations__', {}).items():
            methods['__annotations__'][name_] = tuple(getattr(module, item.name) for item in always_iterable(funcs))

        return super().__new__(meta, name, bases, methods, **kwargs)

    def __init__(cls, name, bases, attrs, **kwargs):
        super(DispatcherMeta, cls).__init__(name, bases, attrs)

        default = kwargs.get('default')

        for name, _ in cls.__dict__.get('__annotations__', {}).items():
            prop = Property(cls.__dict__.get(name, default))
            prop.__set_name__(cls, name)
            setattr(cls, name, prop)

        for name, attr in cls.__dict__.items():
            if name.startswith('__') or isinstance(attr, (FunctionType, Property)):
                continue

            prop = Property(default if isinstance(attr, MISSING) else attr)
            prop.__set_name__(cls, name)
            setattr(cls, name, prop)


class Dispatcher(metaclass=DispatcherMeta):
    def __init_subclass__(cls, **kwargs):
        """Grabbing kwargs"""

    def __init__(self):
        for name, attr in self.__class__.__dict__.items():
            if isinstance(attr, Property):
                self.__dict__[name] = attr.default

        for name, funcs in self.__class__.__annotations__.items():
            self._bind_iter(name, funcs)

    def __getattr__(self, name):
        if name.startswith('bind_to_'):
            return self.__class__.__dict__[name[8:]].bind_deco(self)
        return super().__getattribute__(name)

    def __repr__(self):
        props = ', '.join(f'<{name}={getattr(self, name)}, '
                          f'[{", ".join(callback.__name__ for callback in attr._instance_to_callbacks[self])}]>'
                          for name, attr in self.__class__.__dict__.items() if isinstance(attr, Property))
        return f'{self.__class__.__name__}({props})'

    def _bind_iter(self, name, funcs):
        for func in always_iterable(funcs):
            self.__class__.__dict__[name].bind(self, func)

    def bind(self, **kwargs):
        for key, funcs in kwargs.items():
            if key.startswith('on_'): key = key[3:]
            self._bind_iter(key, funcs)

test_dispatcher.py

from dispatcher import Dispatcher

def observer1(instance, name, old, new):
    print(f'{name} from {instance} changed from {old} to {new}')

def observer2(*args):
    print('Another observer')

class Relay(Dispatcher, default=0):      # bind functions to properties at class definition time
    width: observer1                     # can use default value (default=... in class def)...
    height: (observer1, observer2) = 10  # ...or set initial value
    x                                    # Binding isn't necessary, can still use default...
    y = 10                               # ...or set initial value

relay = Relay()

print(relay.width, relay.height, relay.x, relay.y)  # 0, 10, 0, 10
relay.width = 10  # width from Relay(<height=10, [observer1, observer2]>,
                  #                  <x=0, []>, <y=10, []>, <width=10, [observer1]>) changed from 0 to 10


@relay.bind_to_x                         # bind functions to properties with decorators too...
@relay.bind_to_y
def observer3(instance, name, old, new):
    print(f'{name} changed!')

relay.bind(on_x=lambda *args: print(f'Seriously, {args[1]} changed!')) # ...or with `bind` method

relay.x = 15 # x changed!
             # Seriously, x changed!