-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathobserver.py
74 lines (56 loc) · 1.68 KB
/
observer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from abc import ABC, abstractmethod
from typing import List
# Observer interface
class Observer(ABC):
@abstractmethod
def update(self, price: float):
pass
# Observable
class Subject(ABC):
@abstractmethod
def register_observer(self, observer: Observer):
pass
@abstractmethod
def remove_observer(self, observer: Observer):
pass
@abstractmethod
def notify_observers(self, price: float):
pass
# Concrete Observer
class StockObserver(Observer):
def __init__(self, name: str):
self.name = name
def update(self, price: float):
print(f"{self.name} - Price updated: {price}")
# Concrete Observable
class Stock(Subject):
def __init__(self, symbol: str):
self.symbol = symbol
self.price = 0.0
self.observers: List[Observer] = []
def register_observer(self, observer: Observer):
self.observers.append(observer)
def remove_observer(self, observer: Observer):
self.observers.remove(observer)
def notify_observers(self, price: float):
for observer in self.observers:
observer.update(price)
def set_price(self, price: float):
self.price = price
self.notify_observers(self.price)
# Client code
def client_code():
stock = Stock("AAPL")
observer1 = StockObserver("Observer 1")
observer2 = StockObserver("Observer 2")
stock.register_observer(observer1)
stock.register_observer(observer2)
stock.set_price(150.0)
stock.set_price(155.0)
# Usage
client_code()
## Output
# Observer 1 - Price updated: 150.0
# Observer 2 - Price updated: 150.0
# Observer 1 - Price updated: 155.0
# Observer 2 - Price updated: 155.0