Observer Design Pattern

obersver_design_pattern_example.py

# Observer Design Pattern
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import List, Optional
from random import randrange


class Subject(ABC):
    """
    The Subject interface declares a set of methods for managing and
    notifying subscribers.
    """

    @abstractmethod
    def attach(self, observer: Observer) -> None:
        """
        Attach an observer to the subject.
        """
        pass

    @abstractmethod
    def detach(self, observer: Observer) -> None:
        """
        Detach an observer from the subject.
        """
        pass

    @abstractmethod
    def notify(self) -> None:
        """
        Notify all observers about an event.
        """
        pass


class ConcreteSubject(Subject):
    """
    The Subject owns some important state and notifies observers when the state
    changes.
    """

    # The state is stored in this property for simplicity's sake.
    @property
    def state(self) -> int:
        return self._state

    # Can't modify the state without notifying observers.
    @state.setter
    def state(self, state: int) -> None:
        self._state = state
        print("Subject: My state has just changed to: ", self._state)
        self.notify()

    def __init__(self, observers: Optional[List[Observer]] = None) -> None:
        # List of subscribers.
        # In real life, could be a more complex data structure.
        self._observers: List[Observer] = observers or []

        # Init the state property.
        self._state = 0

    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            print("Subject: Attached an observer.")
            self._observers.append(observer)
        else:
            print("Subject: Observer already attached.")

    def detach(self, observer: Observer) -> None:
        print("Subject: Detaching an observer.")
        self._observers.remove(observer)

    """
    The subscription management methods.
    Note the update() is called in the state setter.
    """

    def notify(self) -> None:
        """
        Trigger an update in each subscriber.
        """

        print("Subject: Notifying observers...")
        for observer in self._observers:
            observer.update(self)

    def some_business_logic(self) -> None:
        """
        Subjects commonly hold more logic that triggers a
        notification whenver something important is about to happen.
        """

        print("\nSubject: I'm doing something important.")
        self.state = randrange(0, 10)


class Observer(ABC):
    """
    The observer interface declares the update method, used by subjects.
    """

    @abstractmethod
    def update(self, subject: Subject) -> None:
        """
        Receive update from subject.
        """
        pass


"""
Concrete Observers react to the updates issued by the Subject they had been attached to.
"""


class ConcreteObserverA(Observer):
    def update(self, subject: Subject) -> None:
        if subject.state < 6:
            print("ConcreteObserverA: Reacted to the event.")


class ConcreteObserverB(Observer):
    def update(self, subject: Subject) -> None:
        if subject.state == 0 or subject.state >= 2:
            print("ConcreteObserverB: Reacted to the event.")


if __name__ == "__main__":
    # The client code.
    subject = ConcreteSubject()
    observerA = ConcreteObserverA()
    observerB = ConcreteObserverB()

    subject.attach(observerA)
    subject.attach(observerB)

    subject.some_business_logic()
    subject.some_business_logic()

    subject.detach(observerB)

    subject.some_business_logic()