Observer パターン

CoddyのPythonジャーニー「オブジェクト指向プログラミング」セクションの一部。レッスン 47/64。

Observer パターンは、あるオブジェクト（subject）の状態が変化したときに、1つのオブジェクト（subject）が複数のオブジェクト（observers）に通知する1対多の関係を作成します。

オブザーバーを管理する単純な Subject クラスを次に示します：

class Subject:
    def __init__(self):
        self._observers = []
    
    def add_observer(self, observer):
        self._observers.append(observer)
    
    def notify(self, message):
        for observer in self._observers:
            observer.update(message)

対象はオブザーバーのリストを保持し、それらすべてに一度に通知できます。

シンプルなオブザーバークラスを作成します：

class EmailNotifier:
    def update(self, message):
        print(f"Email sent: {message}")

class SMSNotifier:
    def update(self, message):
        print(f"SMS sent: {message}")

各オブザーバーには、通知を受けたときに呼び出される update メソッドがあります。

オブザーバーパターンを使用します：

# サブジェクトを作成
news = Subject()

# オブザーバーを作成
email = EmailNotifier()
sms = SMSNotifier()

# サブジェクトにオブザーバーを追加
news.add_observer(email)
news.add_observer(sms)

# すべてのオブザーバーに通知
news.notify("Breaking news: Python is awesome!")

株価トラッカーを使った実践的な例を作成しましょう。

class Stock:
    def __init__(self, symbol, price):
        self.symbol = symbol
        self._price = price
        self._observers = []
    
    def add_observer(self, observer):
        self._observers.append(observer)
    
    def set_price(self, price):
        self._price = price
        self.notify()
    
    def notify(self):
        for observer in self._observers:
            observer.update(self.symbol, self._price)

class Investor:
    def __init__(self, name):
        self.name = name
    
    def update(self, symbol, price):
        print(f"{self.name} notified: {symbol} is now ${price}")

# 株式トラッカーを使用する
apple_stock = Stock("AAPL", 150)

investor1 = Investor("Alice")
investor2 = Investor("Bob")

apple_stock.add_observer(investor1)
apple_stock.add_observer(investor2)

apple_stock.set_price(155)  # すべての投資家に通知する

出力：

Email sent: Breaking news: Python is awesome!
SMS sent: Breaking news: Python is awesome!
Alice notified: AAPL is now $155
Bob notified: AAPL is now $155

重要なポイント：Observer Patternを使うと、何かが変化したときに、1つのオブジェクトから他の多くのオブジェクトへ自動的に通知できます。subjectはobserverのリストを保持し、必要に応じてそれらのupdateメソッドを呼び出します。これは、通知、イベントシステム、アプリケーションの複数の部分の同期を維持する場合に役立ちます。

チャレンジ

中級

天気監視システムを作成して Observer パターンを実装してください。指定されたコード領域に2つのクラスを作成する必要があります。

ステップ1: WeatherStation クラスを作成する

コメントで示されている場所に WeatherStation クラスを記述してください。このクラスは次のようにします。

Subject クラスを継承する（class WeatherStation(Subject): を使用）
次のように初期化する:
- super().__init__() を使用して親のコンストラクターを呼び出す
- 単一アンダースコア属性 self._temperature（内部属性に対する Python の慣例）を使用して、初期温度を 0 に設定する
set_temperature(self, temperature) を実装する:
- 内部の温度属性を更新する
- すべてのオブザーバーに通知するために self.notify(self._temperature) を呼び出す
get_temperature(self) を実装する:
- 現在の温度値を返す

ステップ2: WeatherDisplay クラスを作成する

コメントで示されている場所に WeatherDisplay クラスを記述してください。このクラスは次のようにします。

Observer クラスを継承する（class WeatherDisplay(Observer): を使用）
次のように初期化する:
- name パラメーターを受け取る
- 名前を self.name として保存する
update(self, temperature) を実装する:
- 以下に示す正確な形式で温度更新メッセージを出力する

メッセージ形式:

ディスプレイが温度更新を受け取ったとき、正確に次のように出力する必要があります。

Display [name]: Current temperature is [temperature]C

使用例:

# Create weather station and displays
station = WeatherStation()
phone_display = WeatherDisplay("Phone")
tablet_display = WeatherDisplay("Tablet")

# Attach displays to station
station.attach(phone_display)
station.attach(tablet_display)

# Update temperature - both displays will be notified
station.set_temperature(25.5)

# Output:
# Display Phone: Current temperature is 25.5C
# Display Tablet: Current temperature is 25.5C

重要な注意事項:

attach() メソッドと detach() メソッドは Subject 基底クラスにすでに実装されています。これらを実装したり、WeatherStation クラス内で呼び出したりする必要はありません
上記の使用例では、station.attach(phone_display) は WeatherStation の実装内部ではなく、クラスの利用者によって呼び出されます
WeatherStation では、温度が変化したときに self.notify() を呼び出すだけで構いません。残りの処理は Subject 基底クラスが行います
上記の指定に従い、WeatherStation に必要な3つのメソッドと WeatherDisplay に必要な2つのメソッドの実装に集中してください

自分で試してみよう

class Subject:
    def __init__(self):
        self._observers = []
    
    def attach(self, observer):
        self._observers.append(observer)
        
    def detach(self, observer):
        self._observers.remove(observer)
        
    def notify(self, data):
        for observer in self._observers:
            observer.update(data)
            
class Observer:
    def update(self, data):
        pass

# ここに WeatherStation クラスを書いてください

# ここに WeatherDisplay クラスを書いてください