Skip to content

Instantly share code, notes, and snippets.

@Sean-Bradley
Created August 17, 2020 17:26
Show Gist options
  • Save Sean-Bradley/e3e8b1a9b552d19ed0a5ee0401281af1 to your computer and use it in GitHub Desktop.
Save Sean-Bradley/e3e8b1a9b552d19ed0a5ee0401281af1 to your computer and use it in GitHub Desktop.
"""
Observer Design Pattern
"""
from abc import ABCMeta, abstractmethod
class IObservable(metaclass=ABCMeta):
@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):
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, **kwargs):
for observer in self._observers:
observer.notify(self, *args, **kwargs)
class IObserver(metaclass=ABCMeta):
@staticmethod
@abstractmethod
def notify(observable, *args, **kwargs):
"""Receive notifications"""
class Observer(IObserver):
def __init__(self, observable):
observable.subscribe(self)
def notify(self, observable, *args, **kwargs):
print("Observer received", args, kwargs)
SUBJECT = Subject()
OBSERVERA = Observer(SUBJECT)
OBSERVERB = Observer(SUBJECT)
SUBJECT.notify("Hello Observers")
SUBJECT.unsubscribe(OBSERVERB)
SUBJECT.notify("Hello Observers")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment