observer
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import List
class Subject(ABC):
@abstractmethod
def attach(self, observer: Observer) -> None:
pass
@abstractmethod
def detach(self, observer: Observer) -> None:
pass
@abstractmethod
def notify(self) -> None:
pass
@abstractmethod
def get_state(self) -> int:
pass
class SomeSubject(Subject):
state: int = None
_observers: List[Observer] = []
def attach(self, observer: Observer) -> None:
self._observers.append(observer)
def detach(self, observer: Observer) -> None:
self._observers.remove(observer)
def notify(self) -> None:
for observer in self._observers:
observer.update(self)
def update_state(self, new_state: int) -> None:
self.state = new_state
self.notify()
def get_state(self) -> int:
return self.state
class Observer(ABC):
@abstractmethod
def update(self, subject: Subject) -> None:
pass
class SomeObserverA(Observer):
def update(self, subject: Subject) -> None:
if subject.get_state() < 3:
print(f"SomeObserverA: Reacted to the event, state={subject.get_state()}")
class SomeObserverB(Observer):
def update(self, subject: Subject) -> None:
if subject.get_state() >= 2:
print(f"SomeObserverB: Reacted to the event, state={subject.get_state()}")
if __name__ == "__main__":
subject = SomeSubject()
observer_a = SomeObserverA()
subject.attach(observer_a)
observer_b = SomeObserverB()
subject.attach(observer_b)
subject.update_state(1)
subject.update_state(2)
subject.detach(observer_a)
subject.update_state(3)Last updated