A Python decorator for an "observable property", that notifies observers when the data is changed

Question

I have classes that have some data, and whenever this data changes, it should notify observers of that data change. There also should be a method to add new observers.

My first approach was to use a property with a custom setter. This means that whenever the = operator was used on the data it would call the custom setter, which would notify the observers.

Like so:

# (1)
from typing import Callable

class Foo:

    def __init__(self):
        self.__bar = 0
        self.__bar_observers = []

    @property
    def bar(self):
        return self.__bar

    @bar.setter
    def bar(self, value):
        self.__bar = value
        #notify observers
        for obs in self.__bar_observers:
            obs(value)

    def add_bar_observer(self, observer : Callable[[int], None]):
        self.__bar_observers.append(observer)

f = Foo()
f.add_bar_observer(lambda newval : print(f'new bar val {newval}'))
f.bar = 8

But this is a lot of boilerplate and is many lines of code just for one property. If Foo had 10 observable properties instead of just one, this class would be quite large.

It would be good if this pattern could be created automatically with a decorator.

Here is my attempt at that:

# (2)
def observable_property(initial_value):

    class _observable_property:

        def __init__(self, fget):
            pass

        # We don't know if __set__, __get__, or add_observer will be called first, so we need to call this at the start of all 3.
        def __init(self, obj):
            #On first time called, create the data
            if not hasattr(obj, f'__{self.name}'):
                setattr(obj, f'__{self.name}', initial_value)
            if not hasattr(obj, f'__{self.name}_observers'):
                setattr(obj, f'__{self.name}_observers', [])

        def __get__(self, obj, objtype=None):
            self.__init(obj)
            return getattr(obj, f'__{self.name}')

        def __set__(self, obj, value):
            self.__init(obj)
            setattr(obj, f'__{self.name}', value)
            #Notify all of the observers
            for obs in getattr(obj, f'__{self.name}_observers'):
                obs(value)

        def __set_name__(self, owner: type, name : str):
            self.name = name
            #Add a function to the owner to add new observers
            def add_observer(obj, observer):
                self.__init(obj)
                getattr(obj, f'__{name}_observers').append(observer)

            setattr(owner, f'add_{name}_observer', add_observer)
            

    return _observable_property

And then a test:

# (3)
Meters = NewType('Meters', float)
Kelvin = NewType('Kelvin', float)

class Axis:

    @observable_property(initial_value=Meters(0.0))
    def position(self):
        pass

    @observable_property(initial_value=Kelvin(0.0))
    def temperature(self):
        pass

a = Axis()
a.add_position_observer(lambda newval : print(f'New position for a: {newval}'))
a.add_temperature_observer(lambda newval : print(f'New temperature for a: {newval}'))
a.position = Meters(5.0)
a.temperature = Kelvin(300.0)

This works, but two downsides I can see are that the add_{}_observer functions do not appear in an IDE IntelliSense autocomplete, and it also loses its type hints.

What do you think? Would you use this in production code?

I would also prefer it if the initial_value was set in the constructor of the class, instead of as a decorator parameter, and then the function to be decorated returns that, as is the case with the standard @property decorator.

i.e

# (4)
class Axis:

    def __init__(self):
        self.__position=Meters(0.0)

    @observable_property
    def position(self):
        return self.__position

But I cannot do this as I can't mutate the self.__position from the __set__ descriptor. Is there any way to get the style of # (4), with the functionality of # (3)?

Thanks.

Answer 1

Private name mangling

The reason why you "can't mutate the self.__position from the __set__ descriptor" is the leading double underscore. From 6.2.1. Identifiers (Names):

Private name mangling: When an identifier that textually occurs in a class definition begins with two or more underscore characters and does not end in two or more underscores, it is considered a private name of that class. Private names are transformed to a longer form before code is generated for them. The transformation inserts the class name, with leading underscores removed and a single underscore inserted, in front of the name. For example, the identifier __spam occurring in a class named Ham will be transformed to _Ham__spam. This transformation is independent of the syntactical context in which the identifier is used. If the transformed name is extremely long (longer than 255 characters), implementation defined truncation may happen. If the class name consists only of underscores, no transformation is done.

In the constructor, self.__position = 0.0 is rewritten to be self._Axis__position = 0.0. When you are in the descriptor's __set__ method, getattr(obj, '__position') does not exists, because the name is wrong.

If you used a single leading underscore, instead of a double leading underscore, no name mangling will take place, and attribute can be referenced with the same name in both the main class and the descriptors. This means you can set the initial values in the constructor, as desired.

Boilerplate

    @observable_property(initial_value=Meters(0.0))
    def position(self):
        pass

The def, (self): and pass parts of this are also boilerplate. If you tried to use actual code, instead of the pass statement, the code would never be executed, because the descriptor's __get__ method is used instead.

The above code is approximately:

    position = observable_property(initial_value=Meters(0.0))(lambda self: pass)

If you removed the unused fget argument from the constructor, or removed the _observable_property constructor entirely, this could be reduced to:

    position = observable_property(initial_value=Meters(0.0))()

Since you don't need (or want) the initial value in the observable property descriptor, you want it in the class constructor, and the previous section showed why it wasn't working and how to fix it, we can remove the initial_value argument entirely, and with it the def observable_property wrapper around the class _observable_property.

Reworked code:

from typing import NewType

class ObservableProperty:
    
    def __set_name__(self, owner: type, name: str) -> None:
        def add_observer(obj, observer):
            if not hasattr(obj, self.observers_name):
                setattr(obj, self.observers_name, [])
            getattr(obj, self.observers_name).append(observer)
            
        self.private_name = f'_{name}'
        self.observers_name = f'_{name}_observers'
        setattr(owner, f'add_{name}_observer', add_observer)
    
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name)

    def __set__(self, obj, value):
        setattr(obj, self.private_name, value)
        for observer in getattr(obj, self.observers_name, []):
            observer(value)


Meters = NewType('Meters', float)
Kelvin = NewType('Kelvin', float)

class Axis:

    def __init__(self):
        self._position = Meters(0.0)
        self._temperature = Kelvin(273.0)

    position = ObservableProperty()
    temperature = ObservableProperty()

a = Axis()
print(a.position)     # Initial values from the constructor
print(a.temperature)
a.add_position_observer(lambda newval : print(f'New position for a: {newval}'))
a.add_temperature_observer(lambda newval : print(f'New temperature for a: {newval}'))
a.position = Meters(5.0)
a.temperature = Kelvin(300.0)

Composition

The ObservableProperty descriptor takes away all control of the attribute from the class. The attribute cannot be calculated. For instance, an observable attribute celsius could not read self.kelvin + 273, or write self.kelvin = value - 273 because it has no control over the get/set functionality.

What if you made an @observable decorator, which just added the observer capability, which could be used like:

class Axis:

    def __init__(self):
        self._position = 0.0

    @property
    def position(self) -> Meters:
        return self._position

    @observable
    @position.setter
    def position(self, value: Meters) -> None:
        self._position = value

Then, you could have validation in the setter (such as position can't be negative).

And if you wanted, you could create a SimpleProperty, which had the trivial getter & setter, and compose that into ObservableSimpleProperty and so on.

Answer 2

How about an mixin Observable class.

from collections import defaultdict

class Observable:
    def __init__(self):
        self.observed = defaultdict(list)
    
    def __setattr__(self, name, value):
        super().__setattr__(name, value)
        
        for observer in self.observed.get(name, []):
            observer(value)        
        
    def add_observer(self, name, observer):
        self.observed[name].append(observer)

To be used like so:

class Foo(Observable):
    def __init__(self):
        super().__init__()
        self.aaa = 42

f = Foo()
f.add_observer('aaa', lambda value:print(f"a.aaa changed to {value}"))
f.add_observer('aaa', lambda value:print(f"a.aaa is now {value}"))

Output:

f.aaa = 21
a.aaa changed to 21
a.aaa is now 21