Python Empowerment: Unveiling the Magic of Binding Properties Pattern Implementation for Seamless Code Interactivity

Introduction

In multi-threaded applications it can be necessary to synchronize properties between objects, or at least be notified of changes on certain properties. This is where the Binding Properties comes in, which is basically a form of the Observer pattern.

In this pattern, observers can subscribe to a special ‘event’-handler on an object. When a property changes, all the subscribers, if any, will be notified.

The flow is as follows:

1. An object with bindable properties is created
2. One or more observers subscribe to this object
3. If a property changes in this object, the subscribers get notified and handle the change accordingly.

Implementation in Python

In this example we will build a simple Person class. In our world, a person just has a name and an age. This class has a number, or as the case may be no, observers which will be notified any time a property changes.

Since we are buidling a multi-threaded application, we need to have this import at the start of the file:

import threading

All observers will implement the same class when listening to property-changes:

class Observable:
    def property_changed(self, property_name: str, new_value):
        pass

This class, or we may as well call it an interface, implements one method property_changed() which takes the name of the property which is changed, and the new value of this property.

The Person class

Next we will need to implement the Person class which contains the observable properties.

We start by defining the class, its properties and the constructor:

class Person:
    _name: str = None
    _age: int = None
    _observers: list[Observable] = None
    _lock: threading.Lock = threading.Lock()

    def __init__(self, name: str, age: int) -> None:
        self._name = name
        self._age = age
        self._observers = []

Some explanation:

1. Note that all fields are private. Since the name and the age properties need to be thread-safe we will access them using properties as we will see.
2. We also maintain a list of object implementing the Observable objects, which are notified once a property is changed.
3. The _lock field is used to make sure we can lock our object when needed. We will momentarily see how this is used.

The name and age properties

Python has a very elegant way of defining properties, where we can add extra logic when setting or getting a value. As an example have a look at the name property:

    @property
    def name(self) -> str:
        with self._lock:
            return self._name

    @name.setter
    def name(self, new_value: str) -> None:
        with self._lock:
            self._name = new_value
        self.notify_observers("name", new_value)

Two things to note:

1. Both methods lock the object before changing the value or returning it
2. After setting a new value, we notify the observers.

Since this properties are access like normal instance variables, i.e. person.name=’John’, this logic is completely transparent to the user of this class.

The age property is built along the same lines:

    @property
    def age(self) -> int:
        with self._lock:
            return self._age

    @age.setter
    def age(self, new_value: int) -> None:
        with self._lock:
            self._age = new_value
        self.notify_observers("age", new_value)

Subscribing to changes

We need a thread-safe way to add to our list of observers, which is done in the subscribe() method:

    def subscribe(self, observer: Observable) -> None:
        with self._lock:
            self._observers.append(observer)

And we need a way to notify our observers in case of a state change. This is done by locking the object, and iterating over the list of observers, calling the property_changed() method. The individual observers can decided whether to handle the state change or ignore it:

    def notify_observers(self, property_name: str, new_value) -> None:
        with self._lock:
            for observer in self._observers:
                observer.property_changed(property_name, new_value)

The Observers

Now the time has come to implement our observers. We will start by implementing the NameObserver class:

class NameObserver(Observable):
    def property_changed(self, property_name: str, new_value):
        if property_name == "name":
            print(f"NameObserver: {property_name} changed to {new_value}")

For simplicity’s sake this class has no internal state and just implements the property_changed() method. As you can see only the case where the name property is changed is handled.

The AgeObserver is implemented similarly:

class AgeObserver(Observable):
    def property_changed(self, property_name: str, new_value):
        if property_name == "age":
            print(f"AgeObserver: {property_name} changed to {new_value}")

The change_person()_info function

This function, outside of the Person class, just changes the name and age properties of a Person object, and uses a threading.Event to signal its completion:

def change_person_info(new_person: Person, change_person_event: threading.Event) -> None:
    new_person.name = "Jane"
    new_person.age = 21
    change_person_event.set()

The full code for the Person class:

class Person:
    _name: str = None
    _age: int = None
    _observers: list[Observable] = None
    _lock: threading.Lock = threading.Lock()

    def __init__(self, name: str, age: int) -> None:
        self._name = name
        self._age = age
        self._observers = []

    @property
    def name(self) -> str:
        with self._lock:
            return self._name

    @name.setter
    def name(self, new_value: str) -> None:
        with self._lock:
            self._name = new_value
        self.notify_observers("name", new_value)

    @property
    def age(self) -> int:
        with self._lock:
            return self._age

    @age.setter
    def age(self, new_value: int) -> None:
        with self._lock:
            self._age = new_value
        self.notify_observers("age", new_value)

    def subscribe(self, observer: Observable) -> None:
        with self._lock:
            self._observers.append(observer)

    def notify_observers(self, property_name: str, new_value) -> None:
        with self._lock:
            for observer in self._observers:
                observer.property_changed(property_name, new_value)

Time to test

Now we can test our binding properties:

if __name__ == "__main__":
    person: Person = Person("Test", 55)

    person.subscribe(NameObserver())
    person.subscribe(AgeObserver())

    event: threading.Event = threading.Event()

    threading.Thread(target=change_person_info, args=(person, event)).start()
    event.wait()

    person.name = "John"
    person.age = 55

Line by line:

1. We create a Person instance, and subscribe two observers to it.
2. Then we start a separate thread, and wait for its completion using the wait() on the Event class.
3. Next we change our persons properties back.

The use of the threading.Lock ensures our class properties can be accessed in a thread-safe manner. This is crucial in preventing race conditions.

Conclusion

Implementing this pattern, as with many similar patterns, is quite easy and straightforward in Python also considering we have threading.Lock and threading.Event in the standard library, which makes locking and synchronization quite easy.

One possible enhancement would be to set a different observer to each property, but that is the subject for a next article.