Observer Pattern
Video Lecture
Overview
... Refer to Book, pause Video Lectures or subscribe to Medium Membership to read textual content.
Terminology
... Refer to Book, pause Video Lectures or subscribe to Medium Membership to read textual content.
Observer UML Diagram
Source Code
... Refer to Book, pause Video Lectures or subscribe to Medium Membership to read textual content.
./observer/observer_concept.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65 | # pylint: disable=too-few-public-methods
"Observer Design Pattern Concept"
from abc import ABCMeta, abstractmethod
class IObservable(metaclass=ABCMeta):
"The Subject Interface"
@staticmethod
@abstractmethod
def subscribe(observer):
"The subscribe method"
@staticmethod
@abstractmethod
def unsubscribe(observer):
"The unsubscribe method"
@staticmethod
@abstractmethod
def notify(observer):
"The notify method"
class Subject(IObservable):
"The Subject (Observable)"
def __init__(self):
self._observers = set()
def subscribe(self, observer):
self._observers.add(observer)
def unsubscribe(self, observer):
self._observers.remove(observer)
def notify(self, *args):
for observer in self._observers:
observer.notify(*args)
class IObserver(metaclass=ABCMeta):
"A method for the Observer to implement"
@staticmethod
@abstractmethod
def notify(observable, *args):
"Receive notifications"
class Observer(IObserver):
"The concrete observer"
def __init__(self, observable):
observable.subscribe(self)
def notify(self, *args):
print(f"Observer id:{id(self)} received {args}")
# The Client
SUBJECT = Subject()
OBSERVER_A = Observer(SUBJECT)
OBSERVER_B = Observer(SUBJECT)
SUBJECT.notify("First Notification", [1, 2, 3])
SUBJECT.unsubscribe(OBSERVER_B)
SUBJECT.notify("Second Notification", {"A": 1, "B": 2, "C": 3})
|
Output
| python ./observer/observer_concept.py
Observer id:2084220160272 received ('First Notification', [1, 2, 3])
Observer id:2084220160224 received ('First Notification', [1, 2, 3])
Observer id:2084220160272 received ('Second Notification', {'A': 1, 'B': 2, 'C': 3})
|
Example Use Case
... Refer to Book, pause Video Lectures or subscribe to Medium Membership to read textual content.
Example UML Diagram
Source Code
./observer/client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27 | "Observer Design Pattern Concept"
from data_model import DataModel
from data_controller import DataController
from pie_graph_view import PieGraphView
from bar_graph_view import BarGraphView
from table_view import TableView
# A local data model that the hypothetical external controller updates
DATA_MODEL = DataModel()
# Add some visualisation that use the dataview
PIE_GRAPH_VIEW = PieGraphView(DATA_MODEL)
BAR_GRAPH_VIEW = BarGraphView(DATA_MODEL)
TABLE_VIEW = TableView(DATA_MODEL)
# A hypothetical data controller running in a different process
DATA_CONTROLLER = DataController()
# The hypothetical external data controller updates some data
DATA_CONTROLLER.notify([1, 2, 3])
# Client now removes a local BAR_GRAPH
BAR_GRAPH_VIEW.delete()
# The hypothetical external data controller updates the data again
DATA_CONTROLLER.notify([4, 5, 6])
|
./observer/table_view.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 | "An observer"
from interface_data_view import IDataView
class TableView(IDataView):
"The concrete observer"
def __init__(self, observable):
self._observable = observable
self._id = self._observable.subscribe(self)
def notify(self, data):
print(f"TableView, id:{self._id}")
self.draw(data)
def draw(self, data):
print(f"Drawing a Table view using data:{data}")
def delete(self):
self._observable.unsubscribe(self._id)
|
./observer/bar_graph_view.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 | "An observer"
from interface_data_view import IDataView
class BarGraphView(IDataView):
"The concrete observer"
def __init__(self, observable):
self._observable = observable
self._id = self._observable.subscribe(self)
def notify(self, data):
print(f"BarGraph, id:{self._id}")
self.draw(data)
def draw(self, data):
print(f"Drawing a Bar graph using data:{data}")
def delete(self):
self._observable.unsubscribe(self._id)
|
./observer/pie_graph_view.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 | "An observer"
from interface_data_view import IDataView
class PieGraphView(IDataView):
"The concrete observer"
def __init__(self, observable):
self._observable = observable
self._id = self._observable.subscribe(self)
def notify(self, data):
print(f"PieGraph, id:{self._id}")
self.draw(data)
def draw(self, data):
print(f"Drawing a Pie graph using data:{data}")
def delete(self):
self._observable.unsubscribe(self._id)
|
./observer/interface_data_view.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 | "The Data View interface"
from abc import ABCMeta, abstractmethod
class IDataView(metaclass=ABCMeta):
"A method for the Observer to implement"
@staticmethod
@abstractmethod
def notify(data):
"Receive notifications"
@staticmethod
@abstractmethod
def draw(data):
"Draw the view"
@staticmethod
@abstractmethod
def delete():
"a delete method to remove observer specific resources"
|
./observer/data_model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25 | "A Data Model that observes the Data Controller"
from interface_data_model import IDataModel
from data_controller import DataController
class DataModel(IDataModel):
"A Subject (Observable)"
def __init__(self):
self._observers = {}
self._counter = 0
# subscribing to an external hypothetical data controller
self._data_controller = DataController()
self._data_controller.subscribe(self)
def subscribe(self, observer):
self._counter = self._counter + 1
self._observers[self._counter] = observer
return self._counter
def unsubscribe(self, observer_id):
self._observers.pop(observer_id)
def notify(self, data):
for observer in self._observers:
self._observers[observer].notify(data)
|
./observer/interface_data_model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 | "A Data Model Interface"
from abc import ABCMeta, abstractmethod
class IDataModel(metaclass=ABCMeta):
"A Subject Interface"
@staticmethod
@abstractmethod
def subscribe(observer):
"The subscribe method"
@staticmethod
@abstractmethod
def unsubscribe(observer_id):
"The unsubscribe method"
@staticmethod
@abstractmethod
def notify(data):
"The notify method"
|
./observer/data_controller.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23 | "A Data Controller that is a Subject"
from interface_data_controller import IDataController
class DataController(IDataController):
"A Subject (Observable)"
_observers = set()
def __new__(cls):
return cls
@classmethod
def subscribe(cls, observer):
cls._observers.add(observer)
@classmethod
def unsubscribe(cls, observer):
cls._observers.remove(observer)
@classmethod
def notify(cls, *args):
for observer in cls._observers:
observer.notify(*args)
|
./observer/interface_data_controller.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 | "A Data Controller Interface"
from abc import ABCMeta, abstractmethod
class IDataController(metaclass=ABCMeta):
"A Subject Interface"
@staticmethod
@abstractmethod
def subscribe(observer):
"The subscribe method"
@staticmethod
@abstractmethod
def unsubscribe(observer):
"The unsubscribe method"
@staticmethod
@abstractmethod
def notify(observer):
"The notify method"
|
Output
| python ./observer/client.py
PieGraph, id:1
Drawing a Pie graph using data:[1, 2, 3]
BarGraph, id:2
Drawing a Bar graph using data:[1, 2, 3]
TableView, id:3
Drawing a Table view using data:[1, 2, 3]
PieGraph, id:1
Drawing a Pie graph using data:[4, 5, 6]
TableView, id:3
Drawing a Table view using data:[4, 5, 6]
|
New Coding Concepts
Python Set
A Python Set is similar to a List. Except that the items in the Set are guaranteed to be unique, even if you try to add a duplicate. A set is a good choice for keeping a collection of observables, since the problem of duplicate observables is automatically handled.
A Set can be instantiated pre-filled as {1, 2, 3} surrounded by curly braces or using the keyword set(), verses [] or list() for a List and () or tuple() for a Tuple. It is not the same as a Dictionary, that also uses {}, since the dictionary items are created as key:value pairs. ie {"a": 1, "b": 2, "c": 3}
| PS> python
>>> items = {"yankee", "doodle", "dandy", "doodle"}
>>> items
{'yankee', 'doodle', 'dandy'}
>>> items.add("grandy")
>>> items
{'grandy', 'yankee', 'doodle', 'dandy'}
>>> items.remove("doodle")
>>> items
{'grandy', 'yankee', 'dandy'}
|
Note
If instantiating an empty Set then use my_object = Set() rather than my_object = {} to reduce ambiguity with creating an empty Dictionary.
Summary
... Refer to Book, pause Video Lectures or subscribe to Medium Membership to read textual content.
Paperback 
