🎨

Pythonでデザインパターンを学ぼう (Observer)

2023/04/01に公開

Python

OOP

デザインパターン

tech

Pythonを用いてのGoFの定義した23個のデザインパターンの一つであるObserverパターンの実装方法について解説します。

Observerパターンは、「振る舞い関するデザインパターン」に分類されます。

Observerパターンとは？

observerパターンには主に大きく分けて二つの役割が存在します。

観察対象(Subject)の何らかの状態が変化した際に、Observer(観察者)に通知を行います。

通知を受けたObserver(観察者)は状態変化に応じた処理を行います。
そのためPublish/subscribeモデルとも言えます。

簡単に言えば、「データの更新を複数のオブジェクト（Observer）に通知する」パターンです。

観察対象(Subject)

Observer(観察者)に通知処理を行う。

観察者(Observer)

Subject(観察対象)の状態変化を監視する。
Subjectから受けた通知によって、処理を行う。

Observerパターンのクラス図

抽象的な観察対象(Subject)
抽象的な観察者(Observer)

上記二つは抽象クラス、又はインターフェースです。

観察対象(ConcreteSubject)
観察者(ConcreteObserver)

抽象的な観察対象、抽象的な観察者を継承して、こちら二つに具体的な実装をしていきます。

簡単なObserverパターンのサンプル

観察対象(Subject)が乱数を20回生成する
二つの観察者(Observer)が、観察対象(Subject)の監視を行い、観察対象(Subject)の状態に変化に応じた処理を行う

1つ目の観察者(Observer)は、観察した数を数字で表示する。

DigitObservser: 30

2つ目の観察者(Observer)は、観察した数を*の個数で表示する。

******************************

Subject（観察対象）の実装

まず最初にSubject(観察対象者)の乱数を生成する処理を記述します。
今回は上のクラス図通り、抽象クラスを作成します。

乱数を生成するだけの簡易的な処理なため、具象クラスだけを実装してもかまいませんが、Subject(観察対象者)が増えてきたり、処理内容が複雑してきたりすることから、親クラスである抽象クラスを作成します。

抽象クラスである Subject を作成します。
数字を取得する get_number、処理を実行するexecuteメソッドに関しては、抽象メソッドとして扱い、サブクラスに実装処理を書かせます。

抽象クラスを継承させることで、抽象メソッドを実装させることを強制させます。

それではSubjectを作成してみます。

from abc import ABCMeta, abstractmethod

class Subject(metaclass=ABCMeta):

    def __init__(self) -> None:
        self.__observers: list[Observer] = []

    def add_observer(self, observer: Observer) -> None:
        self.__observers.append(observer)

    def delete_observer(self, observer: Observer) -> None:
        self.__observers.remove(observer)

    def notify_observer(self) -> None:
        for o in self.__observers:
            o.update(self)

    @abstractmethod
    def get_number(self) -> int:
        raise NotImplementedError

    @abstractmethod
    def execute(self) -> None:
        raise NotImplementedError

上記の抽象クラスは__observersというリスト型の変数(配列)を保持しています。(Pythonでは__と入れることがプライベート変数として扱われます)

add_observer

引数にObserverクラスのobserverを取り、通知先であるobserver(観察者)をインスタンス変数である__observersに追加していきます。

delete_observer

引数にObserverクラスのobserverを取り、通知先であるobserver(観察者)ををインスタンス変数である__observersから削除していきます。

notify_observer

通知処理を行います。

__observersのリストをfor-loopで回して、observer(観察者)のオブジェクトを取り出していき、observerオブジェクトのupdateメソッドを実行していくことで、観察者であるObserverに通知処理を行っていきます。

Subject(観察対象)の具体的な実装

それではSubject(観察対象)の具体的な実装を行っていきます。上記で実装した抽象クラスであるSubjectを継承して、抽象メソッドであるget_numberとexecuteメソッドの実装を行っていきます。

それでは実装クラスであるConcreteSubjectRandomNumberを作成していきます。

executeメソッドで乱数の生成を行った後に、抽象クラスで実装されているnotify_observerメソッドを呼び出します。

import random


class ConcreteSubjectRandomNumber(Subject):
    number: int

    def __init__(self):
        super(Subject, self).__init__()
        self.number: int = 0

    def get_number(self) -> int:
        return self.number

    def execute(self) -> None:
        for _ in range(20):
            self.number: int = random.randint(0, 49)
            self.notify_observer()

Observer（観察者）の実装

次に観察対象(Subject)の状態が変化して、通知を受けた際に応じた処理を行うObserver(観察者)の実装を行っていきます。

from abc import ABCMeta, abstractmethod

class Observer(metaclass=ABCMeta):

    @abstractmethod
    def update(self, subject: Subject) -> None:
        raise NotImplementedError

update

引数にSubjectクラスのsubjectを取るようにします。これはSubjectクラスを継承したクラスのインスタンスを引数として受け取るようにします。

Observer(観察者)の具体的な実装

先程作成したObserverクラスを継承し、updateメソッドを実装します。
ここでのサンプルでは、下記の通り二つのObserver(観察者)を作成します。

1つ目の観察者(Observer)は、観察した数を数字で表示
2つ目の観察者(Observer)は、観察した数を*の個数で表示

抽象クラスであるObserverを継承した以下二つのクラスを作成します。

観察した数を数字で表示を行うDegitObserver
「観察した数を「*」の個数で表示」を行うGraphObserver

import time


class DegitObserver(Observer):
    def update(self, subject: Subject):
        print('DigitObservser: {}'.format(
            subject.get_number()
        ))
        time.sleep(0.1)


class GraphObserver(Observer):
    def update(self, subject: Subject):
        count: int = subject.get_number()
        for _ in range(count):
            print('*', end='')
        time.sleep(0.1)

具象クラスである上記二つのupdateメソッドは、ConcreteSubjectRandomNumberのnotify_observerメソッドから呼びだされるようになっています。

Observer(観察者)とSubject(観察対象)を扱う処理

これまでは、ObserverとSubjectの作成をしました。
それではこの両方を呼び出す処理を作成していきましょう。

from subject import Subject, ConcreteSubjectRandomNumber
from observer import Observer, DegitObserver, GraphObserver


def main() -> None:
    # 乱数生成を行うSubject(観察対象者)の作成
    subject: Subject = ConcreteSubjectRandomNumber()

    # 二つのObserverを作成する(観察者)
    degit_observer: Observer = DegitObserver()
    graph_observer: Observer = GraphObserver()

    # それぞれのObserver(観察者)を登録する
    generator.add_observer(degit_observer)
    generator.add_observer(graph_observer)

    # 処理を実行
    # 上記二つのobserverインスタンスに通知が行われる
    subject.execute()

if __name__ == '__main__':
    main()

Observerパターンの処理の内容のおさらいをします。

Subject(観察対象)を作成（上記サンプルなら乱数を生成するConcreteSubjectRandomNumber）
Subjectの状態が変化した際に、notify_observerメソッドを実行(通知処理を行う)
Observer(観察者)インスタンスのupdateメソッドが実行される

Observerパターンの使い道とは

Observerパターンは、実質は一番上でも説明したとおり、Publish/subscribeモデルと言ったほうが正しいです。

Subject(観察対象)が通知処理を行うPublisherの役目を担い、Observer(観察者)が通知を受け取って処理を行うSubscriberの役目を担うと考えることができます。

そのため典型的な使い道としては以下の通りとなると考えられます。

何らかのイベントが発生した際に、それに対応した処理を行う
メーリングリストで特定の購読者にメッセージを送る

他に使い道などがあれば、是非教えてください。

上記のサンプルコードはこちらにまとめてあります