from random import randrange from typing import List, Optional from behavioral.observer.observer import Observer from behavioral.observer.subject import Subject class ConcreteSubject(Subject): """ The Subject owns some important state and notifies observers when the state changes. """ """ For the sake of simplicity, the Subject's state, essential to all subscribers, is stored in this variable. """ _state: Optional[int] = None """ List of subscribers. In real life, the list of subscribers can be stored more comprehensively (categorized by event type, etc.). """ _observers: List[Observer] = [] @property def state(self) -> int: return self._state def attach(self, observer: Observer) -> None: print("Subject: Attached an observer.") self._observers.append(observer) def detach(self, observer: Observer) -> None: self._observers.remove(observer) """ The subscription management methods. """ 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: """ Usually, the subscription logic is only a fraction of what a Subject can really do. Subjects commonly hold some important business logic, that triggers a notification method whenever something important is about to happen (or after it). """ print("\nSubject: I'm doing something important.") self._state = randrange(0, 10) print(f"Subject: My state has just changed to: {self._state}") self.notify()