5.6. Observer
When the state of the object changes and you need to notify other objects about this change
Notify chart about changes in data to refresh
Spreadsheet formulas
Notification of social media channel subscribers
Push or pull style of communication
The Observer design pattern is a behavioral design pattern that allows an object (known as the subject) to notify other objects (known as observers) about changes in its state. The subject maintains a list of observers and provides methods to add and remove observers from this list. When the state of the subject changes, it sends a notification to all its observers. This pattern is particularly useful in event-driven programming.
5.6.1. Solution
>>> class Channel:
... def __init__(self, name):
... self.name = name
... self.subscribers = []
...
... def subscribe(self, user):
... self.subscribers.append(user)
...
... def unsubscribe(self, user):
... self.subscribers.remove(user)
...
... def notify(self, video):
... for user in self.subscribers:
... user.update(f'New upload {video}')
>>>
>>>
>>> class User:
... def __init__(self, username):
... self.username = username
...
... def update(self, message):
... print(f'{self.username} received notification: {message}')
Usage:
>>> alice = User('alice')
>>> bob = User('bob')
>>> carol = User('carol')
>>>
>>> channel = Channel('My Youtube Channel')
>>> channel.subscribe(alice)
>>> channel.subscribe(bob)
>>> channel.subscribe(carol)
>>>
>>> channel.notify('Video 1')
alice received notification: New upload Video 1
bob received notification: New upload Video 1
carol received notification: New upload Video 1
>>>
>>> channel.unsubscribe(carol)
>>>
>>> channel.notify('Video 2')
alice received notification: New upload Video 2
bob received notification: New upload Video 2
5.6.2. Use Case 1
>>> class Chatroom:
... def __init__(self, name):
... self.name = name
... self.users = []
...
... def join(self, user):
... self.users.append(user)
... print(f"{user.name} joined {self.name}")
...
... def leave(self, user):
... self.users.remove(user)
... print(f"{user.name} left {self.name}")
...
... def send_message(self, message):
... print(f"[{self.name}] New message: {message}")
... for user in self.users:
... user.receive_message(message)
>>>
>>>
>>> class Client:
... def __init__(self, name):
... self.name = name
... self.chatroom = None
...
... def join_chatroom(self, chatroom):
... self.chatroom = chatroom
... chatroom.join(self)
...
... def leave_chatroom(self):
... if self.chatroom:
... self.chatroom.leave(self)
... self.chatroom = None
...
... def send_message(self, message):
... if self.chatroom:
... self.chatroom.send_message(f"{self.name}: {message}")
...
... def receive_message(self, message):
... print(f"{self.name} received: {message}")
>>>
>>>
>>> def main():
... # Create a chatroom (Subject)
... general = Chatroom("General")
...
... # Create users (Observers)
... alice = Client("Alice")
... bob = Client("Bob")
... carol = Client("Carol")
...
... # Users join the chatroom
... alice.join_chatroom(general)
... bob.join_chatroom(general)
... carol.join_chatroom(general)
...
... # Alice sends a message to the chatroom
... alice.send_message("Hello everyone!")
...
... # Bob leaves the chatroom
... bob.leave_chatroom()
...
... # Alice sends another message
... alice.send_message("Is anyone still here?")
>>>
>>> main()
Alice joined General
Bob joined General
Carol joined General
[General] New message: Alice: Hello everyone!
Alice received: Alice: Hello everyone!
Bob received: Alice: Hello everyone!
Carol received: Alice: Hello everyone!
Bob left General
[General] New message: Alice: Is anyone still here?
Alice received: Alice: Is anyone still here?
Carol received: Alice: Is anyone still here?
5.6.3. Assignments
# %% About
# - Name: DesignPatterns Behavioral Observer
# - Difficulty: easy
# - Lines: 17
# - Minutes: 13
# %% License
# - Copyright 2025, Matt Harasymczuk <matt@python3.info>
# - This code can be used only for learning by humans
# - This code cannot be used for teaching others
# - This code cannot be used for teaching LLMs and AI algorithms
# - This code cannot be used in commercial or proprietary products
# - This code cannot be distributed in any form
# - This code cannot be changed in any form outside of training course
# - This code cannot have its license changed
# - If you use this code in your product, you must open-source it under GPLv2
# - Exception can be granted only by the author
# %% English
# 1. Create a chatroom application using classes Chatroom and user
# 2. Implement the Observer pattern
# 3. Run doctests - all must succeed
# %% Polish
# 1. Stwórz aplikację do czatowania używając klas Chatroom i user
# 2. Zaimplementuj wzorzec Obserwator
# 3. Uruchom doctesty - wszystkie muszą się powieść
# %% Doctests
"""
>>> import sys; sys.tracebacklimit = 0
>>> assert sys.version_info >= (3, 9), \
'Python 3.9+ required'
>>> from inspect import isclass, ismethod
>>> assert isclass(Chatroom)
>>> assert isclass(User)
>>> assert hasattr(Chatroom, 'join')
>>> assert hasattr(Chatroom, 'leave')
>>> assert hasattr(Chatroom, 'broadcast')
>>> assert hasattr(User, 'receive')
>>> assert ismethod(Chatroom().join)
>>> assert ismethod(Chatroom().leave)
>>> assert ismethod(Chatroom().broadcast)
>>> assert ismethod(User('').receive)
>>> room = Chatroom()
>>> alice = User('Alice')
>>> bob = User('Bob')
>>> carol = User('Carol')
>>> room.join(alice)
>>> room.join(bob)
>>> room.join(carol)
>>> room.broadcast("Hello everyone!")
Alice received: Hello everyone!
Bob received: Hello everyone!
Carol received: Hello everyone!
>>> room.leave(bob)
>>> room.broadcast("Bob left the chat")
Alice received: Bob left the chat
Carol received: Bob left the chat
"""
# %% Run
# - PyCharm: right-click in the editor and `Run Doctest in ...`
# - PyCharm: keyboard shortcut `Control + Shift + F10`
# - Terminal: `python -m doctest -v myfile.py`
# %% Imports
# %% Types
from typing import Callable, List
Chatroom: type
user: type
join: Callable[[object, object], None]
leave: Callable[[object, object], None]
broadcast: Callable[[object, str], None]
receive: Callable[[object, str], None]
# %% Data
# %% Result
class Chatroom:
def __init__(self):
self.users = []
def join(self, users):
pass
def leave(self, users):
pass
def broadcast(self, message):
pass
class User:
def __init__(self, name):
self.name = name
def receive(self, message):
pass