Observer Pattern Blog: Code Snippets

example_2.py

class Subject(object):
	def __init__(self):
		self.observer_list = []

	def register_observer(obs):
		self.observer_list.append(obs)

	def remove_observer(obs):
		if obs in self.observer_list:
			self.observer_list.remove(obs)

	def notify_observers(updated_news):
		for observer in observer_list:
			observer.update(updated_news)

class Observer(object):
	def update(updated_news):
		raise NotImplementedError

class Newspaper(object):
	def display_news():
		raise NotImplementedError

#
# And we integrate them with our `NewsGetterMachine` and new `NewspaperA` and `NewspaperB` classes like so:
#

class NewsGetterMachine(Subject):
	...

	def news_flash():
		news = self.get_news()
		self.notify_observers(news)

	# Just a temporary function to set news for our tests
	def set_news(news):
		self.news = news
		self.news_flash()

class NewspaperA(Newspaper, Observer):
	# Register Newspaper with the NewsGetterMachine
	def __init__(self, news_getter_machine_subject):
		self.subject = news_getter_machine_subject
		news_getter_machine_subject.register_observer(self)

	def update(updated_news):
		# Do stuff specific to NewspaperA
		# like maybe change the news (welcome to the real world)
		...

		self.display_news(updated_news)

	def display_news(updated_news):
		print updated_news

class NewspaperB(Newspaper, Observer):
	# Register Newspaper with the NewsGetterMachine
	def __init__(self, news_getter_machine_subject):
		self.subject = news_getter_machine_subject
		news_getter_machine_subject.register_observer(self)

	def update(updated_news):
		# Do stuff specific to NewspaperB
		...

		self.display_news(updated_news)

	def display_news(updated_news):
		print updated_news