Observer Pattern

CHAPTER Observer Pattern

You know, Norton, I ve been watching you.

Eddie Murphy, Delirious

If you did the object calisthenics exercise from chapter 12, you would have noticed how difficult it is to reduce the number of lines used in certain methods. This is especially difficult if the object is too tightly coupled with a number of other objects; i.e., one object relies too much on its knowledge of the internals of the other objects. As such, methods have too much detail that is not strictly related to the method in question.

Maybe this seems a bit vague at the moment. Allow me to clarify. Say you have a system where users are able to complete challenges, and upon completion they are awarded points, gain general experience points, and also earn skills points for the skills used to complete the task. Every task needs to be evaluated and scored where relevant. Take a moment and think about how you would implement such a system. Once you have written down the basic architecture, look at the example that follows.

What we want is some way to keep track of the user credits earned and spent, the amount of experience the user has accumulated up until this point, as well as the scores that count toward earning specific badges or achievements. You could think of this system as either some sort of habit-building challenge app or any RPG ever created.

task_completer.py class Task(object):

def __init__(self, user, _type):

self.user = user self._type = _type

219

' Wessel Badenhorst 2017

W. Badenhorst, Practical Python Design Patterns, https://doi.org/10.1007/978-1-4842-2680-3_14
CHAPTER OBSERVER PATTERN

def complete(self):

self.user.add_experience(1) self.user.wallet.increase_balance(5)

for badge in self.user.badges:

if self._type == badge._type:

badge.add_points(2)

class User(object):

def __init__(self, wallet):

self.wallet = wallet self.badges = [] self.experience = 0

def add_experience(self, amount):

self.experience += amount

def __str__(self):

return "Wallet\t{}\nExperience\t{}\n+ Badges +\n{}\ n++++++++++++++++".format(

self.wallet,

self.experience,

"\n".join([ str(x) for x in self.badges])

)

class Wallet(object):

def __init__(self):

self.amount = 0

def increase_balance(self, amount):

self.amount += amount

def decrease_balance(self, amount):

self.amount -= amount

def __str__(self):

return str(self.amount)

220
CHAPTER 15 OBSERVER PATTERN

class Badge(object):

def __init__(self, name, _type):

self.points = 0

self.name = name self._type = _type self.awarded = False

def add_points(self, amount):

self.points += amount

if self.points > 3:

self.awarded = True

def __str__(self):

if self.awarded:

award_string = "Earned" else:

award_string = "Unearned"

return "{}: {} [{}]".format(

self.name, award_string, self.points

)

def main():

wallet = Wallet() user = User(wallet)

user.badges.append(Badge("Fun Badge", 1)) user.badges.append(Badge("Bravery Badge", 2)) user.badges.append(Badge("Missing Badge", 3))

tasks = [Task(user, 1), Task(user, 1), Task(user, 3)] for task in tasks:

task.complete()

print(user)

if __name__ == "__main__":

main()

221
CHAPTER 15 OBSERVER PATTERN

In the output, we can see the relevant values added to the wallet, experience, and badges, with the right badge being awarded once the threshold is cleared.

Wallet 15

Experience 3 + Badges +

Fun Badge: Earned [4] Bravery Badge: Unearned [0] Missing Badge: Unearned [2] ++++++++++++++++

This very basic implementation has a fairly complex set of calculations to perform whenever a task is completed. In the preceding code, we have a fairly well-implemented architecture, but the evaluation function is still clunky, and I m sure you already get

that feeling that this is code you would not like to work on once you got it working. I also think that it would not be any fun to write tests for this method. Any additions

to the system would mean altering this method, forcing even more calculations and evaluations to take place.

As we stated earlier, this is a symptom of having a tightly coupled system. The Task object must know about every points object in order for it to be able to assign the correct points or credits to the correct sub-system. We want to remove the evaluation of every rule from the main part of the task complete method and place more responsibility

with the sub-systems so they handle data alterations based on their own rules and not those of some foresight object.

To do this, we take the first step toward a more decoupled system, as seen here:

task_semi_decoupled.py class Task(object):

def __init__(self, user, _type):

self.user = user self._type = _type

def complete(self): self.user.complete_task(self)

self.user.wallet.complete_task(self) for badge in self.user.badges: badge.complete_task(self)

222
CHAPTER OBSERVER PATTERN

class User(object):

def __init__(self, wallet):

self.wallet = wallet self.badges = [] self.experience = 0

def add_experience(self, amount):

self.experience += amount

def complete_task(self, task):

self.add_experience(1)

def __str__(self):

return "Wallet\t{}\nExperience\t{}\n+ Badges +\n{}\ n++++++++++++++++".format(

self.wallet,

self.experience,

"\n".join([ str(x) for x in self.badges])

)

class Wallet(object):

def __init__(self):

self.amount = 0

def increase_balance(self, amount):

self.amount += amount

def decrease_balance(self, amount):

self.amount -= amount

def complete_task(self, task): self.increase_balance(5)

def __str__(self):

return str(self.amount)

223
CHAPTER OBSERVER PATTERN

class Badge(object):

def __init__(self, name, _type):

self.points = 0

self.name = name self._type = _type self.awarded = False

def add_points(self, amount):

self.points += amount

if self.points > 3:

self.awarded = True

def complete_task(self, task): if task._type == self._type:

self.add_points(2)

def __str__(self):

if self.awarded:

award_string = "Earned" else:

award_string = "Unearned"

return "{}: {} [{}]".format(

self.name, award_string, self.points

)

def main():

wallet = Wallet() user = User(wallet)

user.badges.append(Badge("Fun Badge", 1)) user.badges.append(Badge("Bravery Badge", 2)) user.badges.append(Badge("Missing Badge", 3))

224
CHAPTER OBSERVER PATTERN

tasks = [Task(user, 1), Task(user, 1), Task(user, 3)] for task in tasks:

task.complete()

print(user)

if __name__ == "__main__":

main()

This results in the same output as before. This is already a much better solution. The evaluation now takes place in the objects where they are relevant, which is closer to the rules contained in the objects callisthenics exercise. I hope a little light went on for you in terms of the value of practicing these types of code callisthenics and how it serves to make you a better programmer.

It still bothers me that the Task handler is changed whenever a new type of badge or credit or alert or whatever else you can think of is added to the system. What would be ideal is if there were some sort of hooking mechanism that would not only allow you to register new sub-systems and then dynamically have their evaluations done as needed, but also free you from the requirement to make any alterations to the main task system.

The concept of callback functions is quite useful for achieving this new level

of dynamics. We will add generic callback functions to a collection that will be run whenever a task is completed. Now the system is more dynamic, as these systems can be added at runtime. The objects in the system will be decoupled even more.

task_with_callbacks.py class Task(object):

def __init__(self, user, _type):

self.user = user

self._type = _type

self.callbacks = [

self.user,

self.user.wallet,

] self.callbacks.extend(self.user.badges)

def complete(self):

for item in self.callbacks:

item.complete_task(self)

225
CHAPTER OBSERVER PATTERN

class User(object):

def __init__(self, wallet):

self.wallet = wallet self.badges = [] self.experience = 0

def add_experience(self, amount):

self.experience += amount

def complete_task(self, task):

self.add_experience(1)

def __str__(self):

return "Wallet\t{}\nExperience\t{}\n+ Badges +\n{}\n++++++++++++++++".format(

self.wallet,

self.experience,

"\n".join([ str(x) for x in self.badges])

)

class Wallet(object):

def __init__(self):

self.amount = 0

def increase_balance(self, amount):

self.amount += amount

def decrease_balance(self, amount):

self.amount -= amount

def complete_task(self, task): self.increase_balance(5)

def __str__(self):

return str(self.amount)

226
CHAPTER OBSERVER PATTERN

class Badge(object):

def __init__(self, name, _type):

self.points = 0

self.name = name self._type = _type self.awarded = False

def add_points(self, amount):

self.points += amount

if self.points > 3:

self.awarded = True

def complete_task(self, task): if task._type == self._type:

self.add_points(2)

def __str__(self):

if self.awarded:

award_string = "Earned" else:

award_string = "Unearned"

return "{}: {} [{}]".format(

self.name, award_string, self.points

)

def main():

wallet = Wallet() user = User(wallet)

user.badges.append(Badge("Fun Badge", 1)) user.badges.append(Badge("Bravery Badge", 2)) user.badges.append(Badge("Missing Badge", 3))

227
CHAPTER OBSERVER PATTERN

tasks = [Task(user, 1), Task(user, 1), Task(user, 3)] for task in tasks:

task.complete()

print(user)

if __name__ == "__main__":

main()

Now you have a list of objects to be called back when the task is completed, and

the task need not know anything more about the objects in the callbacks list other than

that they have a complete_task() method that takes the task that just completed as a parameter.

Whenever you want to dynamically decouple the source of a call from the called

code, this is the way to go. This problem is another one of those very common problems, so common that this is another design pattern the observer pattern. If you take a step back from the problem we are looking at and just think of the observer pattern in general terms, it will look something like this.

There are two types of objects in the pattern, an Observable class, which can be watched by other classes, and an Observer class, which will be alerted whenever an Observable object the two classes are connected to undergoes a change.

In the original Gang of Four book, the observer design pattern is defined as follows:

A software design pattern in which an object, called the subject, maintains a list of its dependents, called observers, and notifies them automatically of any state changes, usually by calling one of their methods. It is mainly used to implement distributed event handling systems [Gamma, E.; Helm, R.; Johnson, R.; Vlissides, J.; Design Patterns: Elements of Reusable Object-Oriented Software Publisher: Addison-Wesley, 1994].

In code, it looks like this:

import abc

class Observer(object):

__metaclass__ = abc.ABCMeta

@abc.abstractmethod

def update(self, observed): pass

228
CHAPTER OBSERVER PATTERN

class ConcreteObserver(Observer):

def update(self, observed):

print("Observing: " + observed)

class Observable(object):

def __init__(self):

self.observers = set()

def register(self, observer):

self.observers.add(observer)

def unregister(self, observer):

self.observers.discard(observer)

def unregister_all(self):

self.observers = set()

def update_all(self):

for observer in self.observers:

observer.update(self)

The use of the Abstract base class results in a Java-like interface that forces the observers to implement the update() method. As we have seen before, there is no need for the Abstract base class, because of Python s dynamic nature, and as such the preceding code can be replaced by a more pythonic version:

class ConcreteObserver(object):

def update(self, observed):

print("Observing: " + observed)

class Observable(object):

def __init__(self):

self.observers = set()

def register(self, observer):

self.observers.add(observer)

229
CHAPTER OBSERVER PATTERN

def unregister(self, observer):

self.observers.discard(observer)

def unregister_all(self):

self.observers = set()

def update_all(self):

for observer in self.observers:

observer.update(self)

In the preceding code, the Observable keeps a record of all the objects observing it in a list called observers, and whenever relevant changes happen in the Observable, it simply runs the update() method for each observer. You will notice this every time the update() function is called with the Observable object passed as a parameter. This is the general way of doing it, but any parameters, or even no parameters, can be passed to the observers, and the code will still adhere to the observer pattern.

If we were able to send different parameters to different objects, that would be really interesting, since we would greatly increase the efficiency of the calls if we no longer needed to pass the whole object that has undergone changes. Let s use Python s dynamic nature to make our observer pattern even better. For this, we will return to our user task system.

general_observer.py

class ConcreteObserver(object):

def update(self, observed):

print("Observing: {}".format(observed))

class Observable(object):

def __init__(self):

self.callbacks = set()

def register(self, callback):

self.callbacks.add(callback)

def unregister(self, callback):

self.callbacks.discard(callback)

230
CHAPTER OBSERVER PATTERN

def unregister_all(self):

self.callbacks = set()

def update_all(self):

for callback in self.callbacks:

callback(self)

def main():

observed = Observable() observer1 = ConcreteObserver()

observed.register(lambda x: observer1.update(x)) observed.update_all()

if __name__ == "__main__":

main()

Although there are many ways to string up the actions that can take place once the state of an Observable changes, these are two concrete examples to build on.

Sometimes, you might find that you would prefer the rest of the system be updated

at a specific time and not whenever the object changes. To facilitate this requirement,

we will add a changed flag in a protected variable (remember, Python does not

explicitly block access to this variable; it is more a matter of convention), which can

be set and unset as needed. The timed function will then only process the change and alert the observers of the Observable object at the desired time. You could use any implementation of the observer pattern combined with the flag. In the example that follows, I used the functional observer. As an exercise, implement the flag for changed on the example with the object set of observers.

import time

class ConcreteObserver(object):

def update(self, observed):

print("Observing: {}".format(observed))

231
CHAPTER OBSERVER PATTERN

class Observable(object):

def __init__(self):

self.callbacks = set() self.changed = False

def register(self, callback):

self.callbacks.add(callback)

def unregister(self, callback):

self.callbacks.discard(callback)

def unregister_all(self):

self.callbacks = set()

def poll_for_change(self): if self.changed: self.update_all

def update_all(self):

for callback in self.callbacks:

callback(self)

def main():

observed = Observable() observer1 = ConcreteObserver()

observed.register(lambda x: observer1.update(x))

while True:

time.sleep(3) observed.poll_for_change()

if __name__ == "__main__":

main()

The problems that the observer pattern solves are those where a group of objects has to respond to the change of state in some other object and do so without causing more coupling inside the system. In this sense, the observer pattern is concerned with the management of events or responding to change of state in some sort of network of objects.

232
CHAPTER OBSERVER PATTERN

I mentioned coupling, so let me clarify what is meant when we talk about coupling. Generally, when we talk about the level of coupling between objects, we refer to the degree of knowledge that one object needs with regard to other objects that it interacts with. The more loosely objects are coupled, the less knowledge they have about each other, and the more flexible the object-oriented system is. Loosely coupled systems have fewer interdependencies between objects and as such are easier to update and maintain. By decreasing the coupling between objects, you further reduce the risk that changing something in one part of the code will have unintended consequences in some other

part of the code. Since objects do not rely on each other, unit testing and troubleshooting become easier to do.

Other good places to use the observer pattern include the traditional Model-View- Controller design pattern, which you will encounter later in the book, as well as a textual description of data that needs to be updated whenever the underlying data changes.

As a rule, whenever you have a publish subscribe relationship between a single object (the Observable) and a set of observers, you have a good candidate for the observer pattern. Some other examples of this type of architecture are the many different types of feeds you find online, like newsfeeds, Atom, RSS, and podcasts. With all of these, the publisher does not care who the subscribers are, and thus you have a natural separation of concern. The observer pattern will make it easy to add or remove subscribers at runtime by increasing the level of decoupling between the subscribers and the publisher.

Now, let s use the pythonic implementation of the observer pattern to implement our tracking system from the beginning of the chapter. Note how easy it is to add new classes to the code and how clean the processes of updating and changing any of the existing classes are.

class Task(object):

def __init__(self, user, _type):

self.observers = set() self.user = user self._type = _type

def register(self, observer):

self.observers.add(observer)

233
CHAPTER OBSERVER PATTERN

def unregister(self, observer):

self.observers.discard(observer)

def unregister_all(self):

self.observers = set()

def update_all(self):

for observer in self.observers:

observer.update(self)

class User(object):

def __init__(self, wallet):

self.wallet = wallet self.badges = [] self.experience = 0

def add_experience(self, amount):

self.experience += amount

def update(self, observed):

self.add_experience(1)

def __str__(self):

return "Wallet\t{}\nExperience\t{}\n+ Badges +\n{}\n++++++++++++++++".format(

self.wallet,

self.experience,

"\n".join([ str(x) for x in self.badges])

)

class Wallet(object):

def __init__(self):

self.amount = 0

def increase_balance(self, amount):

self.amount += amount

234
CHAPTER OBSERVER PATTERN

def decrease_balance(self, amount):

self.amount -= amount

def update(self, observed):

self.increase_balance(5)

def __str__(self):

return str(self.amount)

class Badge(object):

def __init__(self, name, _type):

self.points = 0

self.name = name self._type = _type self.awarded = False

def add_points(self, amount):

self.points += amount

if self.points > 3:

self.awarded = True

def update(self, observed):

if observed._type == self._type:

self.add_points(2)

def __str__(self):

if self.awarded:

award_string = "Earned" else:

award_string = "Unearned"

return "{}: {} [{}]".format(

self.name, award_string, self.points

)

235
CHAPTER OBSERVER PATTERN

def main():

wallet = Wallet() user = User(wallet)

badges = [

Badge("Fun Badge", 1), Badge("Bravery Badge", 2), Badge("Missing Badge", 3)

]

user.badges.extend(badges)

tasks = [Task(user, 1), Task(user, 1), Task(user, 3)]

for task in tasks:

task.register(wallet) task.register(user)

for badge in badges:

task.register(badge)

for task in tasks:

task.update_all()

print(user)

if __name__ == "__main__":

main()

Parting Shots

You should have a clear understanding of the observer pattern by now, and also have a feeling for how implementing this pattern in the real world allows you to build systems that are more robust, easier to maintain, and a joy to extend.

236
CHAPTER OBSERVER PATTERN

Exercises

Use the observer pattern to model a system where your observers can

subscribe to stocks in a stock market and make buy/sell decisions based on changes in the stock price.

Implement the flag for changed on an example with the object set of

observers.

237