You know, Norton, I've been watching you.
(あのな、ノートン、おれはずっとあんたを見張ってたんだぜ)
- Eddie Murphy, Delirious
コードのリファクタリング等を計画している際に、ある特定のメソッドについて書き直しによるコンパクト化などが非常に困難であることに気付く場合があります。この現象は、対象のオブジェクトが他のオブジェクトと密に結合している場合に顕著に現れます。つまり、そのオブジェクトが他のオブジェクトの内部機能にあまりにも頼り過ぎている場合です。しかし逆に、それらのメソッドは様々な機能をごちゃごちゃと詰め込んでいて、対象のメソッドとだけ厳密な関係を築いているわけではないでしょう。
ちょっと話の内容がピンと来ないかもしれませんね。何とかもう少しハッキリさせてみましょう。例えば次のようなシステムがあるとします。ユーザーには課題が与えられ、それをクリアするとコインを取得できます。また経験値ポイントと、その課題を解くために使用したスキルに対するポイントも獲得でき、このポイントをある一定以上貯めるとスキルに応じたバッジを獲得できるようになっています。
しばらく時間をかけてそのようなシステムをどのように実装できるかを考えてみてください。そして基本的な構成を書き出せたら次の例を見てください。
このシステムで必要としているのは、ユーザーが獲得した or 消費したコイン、これまでに積み上げてきた経験、あるバッジを獲得するためのスキルごとのスコア、といったものを追跡する機能です。ですから、ある種の「趣味やスポーツの目標達成監視アプリ」や「ロールプレイングゲーム (RPG)」と考えて差し支えありません:
class Wallet:
def __init__(self):
self.amount: int = 0
def increase_balance(self, amount: int):
self.amount += amount
def decrease_balance(self, amount: int):
self.amount -= amount
def __str__(self) -> str:
return str(self.amount)
class Badge:
def __init__(self, name: str, _type: int):
self.points: int = 0
self.name = name
self._type = _type
self.awarded: bool = False
def add_points(self, amount: int):
self.points += amount
if self.points > 3:
self.awarded = True
def __str__(self) -> str:
if self.awarded:
award_string = "Earned"
else:
award_string = "Unearned"
return f'{self.name:>13}: {award_string} [{self.points}]'
class User:
def __init__(self, wallet: Wallet):
self.wallet = wallet
self.badges: list[Badge] = []
self.experience: int = 0
def add_experience(self, amount: int):
self.experience += amount
def __str__(self) -> str:
NL: str = '\n'
MAX_LEN: int = len(max(['Wallet', 'Experience'], key=lambda x: len(x)))
return (f'{"Wallet":>{MAX_LEN}}:{str(self.wallet):>3}\n'
f'Experience:{self.experience:>3}\n'
f'\n++ Badges ++\n{NL.join([str(x) for x in self.badges])}'
f'\n+++++++++++++++++++++++++++++')
class Task:
def __init__(self, user: User, _type: int):
self.user = user
self._type = _type
def complete(self):
self.user.add_experience(1)
self.user.wallet.increase_balance(5)
for badge in self.user.badges:
if self._type == badge._type:
badge.add_points(2)
def main():
wallet = Wallet()
user = User(wallet)
user.badges.append(Badge('Fun Badge', 1))
user.badges.append(Badge('Bravery Badge', 2))
user.badges.append(Badge('Missing Badge', 3))
tasks = [Task(user, 1), Task(user, 2), Task(user, 3), Task(user, 2)]
for task in tasks:
task.complete()
print(user)
if __name__ == "__main__":
main()
実行結果は次のようになります。ユーザーが現在保持しているコイン、経験値、利用したそれぞれのスキルに対応したバッジ毎のポイント、そして、それぞれのバッジの所有状況が分かります:
Wallet: 20
Experience: 4
++ Badges ++
Fun Badge: ---- [2]
Bravery Badge: 所有 [4]
Missing Badge: ---- [2]
+++++++++++++++++++++++++++++
この非常に基本的なコードはユーザーがタスクを完了するたびに非常に煩雑な計算処理を行っています。プログラム全体の実装はかなり適切に行われていると思えますが、それでもこの評価関数 (Task.complete()) 自体は「こなれている」とは言い難いものです。一旦動作するようになったらもうあまり手を出したくない類のコードだな、とあなた自身も感じているのではないでしょうか？私としてもこのメソッドのためのテストを自ら進んで書きたいとは思いません。このプログラムへの機能の追加は、すなわち、この評価関数の変更を意味します。新しい機能に対する評価を行い、ポイントを計算する処理を追加する必要があります。
この「症状」の原因は、この章の始めでお話しした「密な結合」によるものです。Task.complete() はタスク完了に伴う評価とポイント付与を正確に実行するために、ポイントに関わる全てのサブシステムの詳細を把握しておく必要があります。
Task.complete() と他のポイントオブジェクト間の結合を密ではなくすために、この関数から各ルールに則った評価ルーチンを削除し、それぞれのポイントサブシステムへその処理を受け持たせる必要があります。つまり、それぞれのルールに則った評価はそれぞれのポイントオブジェクトが行い、それに伴うポイント付与等に関しても責任を持つようにする必要があります。
こうした「より疎な結合 (more decoupled)」のシステムを構築するためのファーストステップとして、以下のような実装に変更してみました:
class Wallet:
def __init__(self):
self.amount: int = 0
def increase_balance(self, amount: int):
self.amount += amount
def decrease_balance(self, amount: int):
self.amount -= amount
def complete_task(self, task: 'Task'):
self.increase_balance(5)
def __str__(self) -> str:
return str(self.amount)
class Badge:
def __init__(self, name: str, _type: int):
self.points: int = 0
self.name = name
self._type = _type
self.awarded: bool = False
def add_points(self, amount: int):
self.points += amount
if self.points > 3:
self.awarded = True
def complete_task(self, task: 'Task'):
if task._type == self._type:
self.add_points(2)
def __str__(self) -> str:
if self.awarded:
award_string = "所有"
else:
award_string = "----"
return f'{self.name:>13}: {award_string} [{self.points}]'
class User:
def __init__(self, wallet: Wallet):
self.wallet = wallet
self.badges: list[Badge] = []
self.experience: int = 0
def add_experience(self, amount: int):
self.experience += amount
def complete_task(self, task: 'Task'):
self.add_experience(1)
def __str__(self) -> str:
NL: str = '\n'
MAX_LEN: int = len(max(['Wallet', 'Experience'], key=lambda x: len(x)))
return (f'{"Wallet":>{MAX_LEN}}:{str(self.wallet):>3}\n'
f'{"Experience":>{MAX_LEN}}:{self.experience:>3}\n'
f'\n++ Badges ++\n{NL.join([str(x) for x in self.badges])}'
f'\n+++++++++++++++++++++++++++++')
class Task:
def __init__(self, user: User, _type: int):
self.user = user
self._type = _type
def complete(self):
self.user.complete_task(self)
self.user.wallet.complete_task(self)
for badge in self.user.badges:
badge.complete_task(self)
def main():
wallet = Wallet()
user = User(wallet)
user.badges.append(Badge('Fun Badge', 1))
user.badges.append(Badge('Bravery Badge', 2))
user.badges.append(Badge('Missing Badge', 3))
tasks = [Task(user, 1), Task(user, 2), Task(user, 3), Task(user, 2)]
for task in tasks:
task.complete()
print(user)
if __name__ == "__main__":
main()
出力結果に変更はありません。しかしシステムのアーキテクチャにはかなりの改善が見られます。タスクの評価処理はそれぞれのルールを規定しているオブジェクトの担当になっています。こういったコードの見直し、書き換えの練習が、より良いプログラマーになるために如何に役立つか、ということに気付くきっかけの１つになってもらえれば、とも思います。
さて、システムはかなりスッキリしたとはいえまだ気にくわないところがあります。それは、新しいバッジであるとか、ポイントに結び付くような何か新しい機能をこのシステムに追加しようとした場合、どうしても Task() に手を加える必要がある、ということです。理想を言えば、サブシステムを新たに登録し、必要に応じて評価処理を行えるようなフックメカニズム (hooking mechanism) が存在すれば最高です。そして、そのメカニズムのおかげでサブシステムの新規追加時にも Task() に変更を加える必要がないのであれば、もう言うことはありません。
このような動的機構を達成するために非常に役に立つのが「コールバック関数 (callback functions」という概念です。タスクが完了した時点で呼び出されるそれぞれのサブシステムに紐付いたコールバック関数のリストを保持するようにしておけば、プログラムの実行時にそれらを変更、追加することでシステム自体もより動的になり、それぞれのオブジェクト間の結び付きもより「疎」にすることができます:
class EvaluationBase:
def complete_task(self, task: 'Task'):
pass
class Wallet(EvaluationBase):
def __init__(self):
self.amount: int = 0
def increase_balance(self, amount: int):
self.amount += amount
def decrease_balance(self, amount: int):
self.amount -= amount
def complete_task(self, task: 'Task'):
self.increase_balance(5)
def __str__(self) -> str:
return str(self.amount)
class Badge(EvaluationBase):
def __init__(self, name: str, _type: int):
self.points: int = 0
self.name = name
self._type = _type
self.awarded: bool = False
def add_points(self, amount: int):
self.points += amount
if self.points > 3:
self.awarded = True
def complete_task(self, task: 'Task'):
if task._type == self._type:
self.add_points(2)
def __str__(self) -> str:
if self.awarded:
award_string = "所有"
else:
award_string = "----"
return f'{self.name:>13}: {award_string} [{self.points}]'
class User(EvaluationBase):
def __init__(self, wallet: Wallet):
self.wallet = wallet
self.badges: list[Badge] = []
self.experience: int = 0
def add_experience(self, amount: int):
self.experience += amount
def complete_task(self, task: 'Task'):
self.add_experience(1)
def __str__(self) -> str:
NL: str = '\n'
MAX_LEN: int = len(max(['Wallet', 'Experience'], key=lambda x: len(x)))
return (f'{"Wallet":>{MAX_LEN}}:{str(self.wallet):>3}\n'
f'{"Experience":>{MAX_LEN}}:{self.experience:>3}\n'
f'\n++ Badges ++\n{NL.join([str(x) for x in self.badges])}'
f'\n+++++++++++++++++++++++++++++')
class Task:
def __init__(self, user: User, _type: int):
self.user = user
self._type = _type
self.callbacks: list[EvaluationBase] = [
self.user,
self.user.wallet,
]
self.callbacks.extend(self.user.badges)
def complete(self):
for item in self.callbacks:
item.complete_task(self)
def main():
wallet = Wallet()
user = User(wallet)
user.badges.append(Badge('Fun Badge', 1))
user.badges.append(Badge('Bravery Badge', 2))
user.badges.append(Badge('Missing Badge', 3))
tasks = [Task(user, 1), Task(user, 2), Task(user, 3), Task(user, 2)]
for task in tasks:
task.complete()
print(user)
if __name__ == "__main__":
main()
この実装で変更したのは Task() メソッドにタスク完了時に呼び出すコールバック関数 (を含むクラス) のリストを追加したことだけです (タイプヒントをつけるために、各評価サブクラスに継承される抽象インターフェースクラスも追加しましたが doc ストリングにも記述したように本来必要ありません)。この変更によって、このリストに含まれるオブジェクトの詳細を Task() は１つの事実を除いて一切知る必要がなくなります。それは、これらのクラスには完了したばかりの Task をパラメータとして受け取る complete_task() メソッドが定義されている、ということです。
このアプローチは、呼び出し元のコードを呼び出される側のコードと切り離す最善の方法です。この問題も非常に一般的なものであることから、この問題に対処するためのこのアプローチもデザインパターンとして確立しています; それがオブザーバーパターン (observer pattern) です。
Gang of Four の原作ではオブザーバーパターンを次のように定義しています:
Subject (サブジェクト) と呼ばれるオブジェクトが Observer (オブザーバー) と呼ばれる依存関係にあるオブジェクトのリストを保持し、ある状態の変化 (タスクが終了した等) が発生した場合に、オブザーバーオブジェクトに一律で定義されているメソッドを呼び出すことで自動的にその変化を通知するソフトウェアデザインパターン。主に分散イベント処理システム (distributed event handling systems) を実装するために使用する。
- Gamma, E.; Helm, R.; Johnson, R.; Vlissides, J.; Design Patterns: Elements of Reusable Object-Oriented Software Publisher: Addison-Wesley, 1994 -
import abc
class Observer:
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def update(self, observed: 'Subject'):
pass
class ConcreteObserver(Observer):
def update(self, observed: 'Subject'):
print(f'通知あり from {observed}')
class Subject:
def __init__(self):
self.observers = set()
def register(self, observer: Observer):
self.observers.add(observer)
def unregister(self, observer: Observer):
self.observers.discard(observer)
def unregister_all(self):
self.observers = set()
def update_all(self):
for observer in self.observers:
observer.update(self)
abc (Abstract Base Classes) ライブラリを利用して Java 言語などにおけるインターフェースを定義することで、それを継承する具象 Observer クラスでは update() メソッドの実装が義務付けられます。ただしこれまでも何回もお話ししているように、Python のダックタイピングにより抽象ベースクラスの定義は必要ありません。つまり上のコードはより pythonic なものに置き換えが可能です:
class ConcreteObserver:
def update(self, observed: 'Subject'):
print(f'通知あり from {observed}')
class Subject:
def __init__(self):
self.observers = set()
def register(self, observer):
self.observers.add(observer)
def unregister(self, observer):
self.observers.discard(observer)
def unregister_all(self):
self.observers = set()
def update_all(self):
for observer in self.observers:
observer.update(self)
Doc ストリングにも記した通りこの実装では、Subject (Observable) オブジェクトで自分を「監視している」Observer クラスの全てのオブジェクトを set に保持しています (よって重複登録はできません)。そして何かしらの状態変化が Subject 自身に発生した場合、登録されているすべての Observer オブジェクトの update() を呼び出しているだけです。そしてこの時に自分自身を渡すことで、状態変化の内容を Observer オブジェクト側が参照できるようにしています。これがオブザーバーパターンの一般的な実装ですが、引数として他の値を渡したり、または、引数を全く渡さなかったとしても、それはそれでオブザーバーパターンであることに変わりはありません。
また、もし引数として様々な値を渡せるようにしておけば、変化が生じたオブジェクト全体を常に渡さなければならないよりも遥かに効率の良い実装となるはずですから非常に興味深いですね。Python の「動的型付け」という性質を利用して更に改良を施してみましょう:
from typing import Callable, Any
class ConcreteObserver:
def update(self, observed: Any):
print(f'通知あり from {observed}')
class Subject:
def __init__(self):
self.callbacks: set[Callable[[Any], Any]] = set()
def register(self, callback: Callable[[Any], Any]):
self.callbacks.add(callback)
def unregister(self, callback: Callable[[Any], Any]):
self.callbacks.discard(callback)
def unregister_all(self):
self.callbacks = set()
def update_all(self):
for callback in self.callbacks:
callback(self)
def main():
observed = Subject()
observer1 = ConcreteObserver()
observed.register(lambda x: observer1.update(x))
observed.update_all()
if __name__ == '__main__':
main()
「監視されている側 (Subject)」の状態の変化と実行されるべき「アクション」を結び付けるための方法はここで紹介したもの以外にも多く存在しますが、オブザーバーパターンを採用した典型的な実装例は以上の２つになります。
時には、監視対象の状態が変化したら「常に」通知を受けてアクションを起こすのではなく「ある特定の時だけ」通知を受けてアクションを起こしたい場合もあるでしょう。この要求を満たすために、Protected 変数としてある条件を満たしたときだけセットするフラグを定義しておくこともできます (Python では private や protected といった「実際に」アクセス制御を行うための機構は用意されていません。これはあくまでも他のプログラマーに対する「宣言」であり「慣習」に過ぎません)。このフラグはある条件を満たした場合のみセットされ、状態変化が生じたとしてもこのフラグがセットされている場合のみ Observers に通知を行うわけです。オブザーバーパターンとフラグをどのように組み合わせて実装するのかはあなたの裁量次第です。
例えば、前回の通知からある一定の時間以上経過していない場合 (この例では 10 秒) は通知を行いたくない場合は次のように記述できるでしょう:
from typing import Callable, Any
from datetime import datetime, timedelta
import time
class ConcreteObserver:
def update(self, observed: Any):
print(f'通知あり at {observed.prev_access_time}\n====================')
class Subject:
def __init__(self):
self.callbacks: set[Callable[[Any], Any]] = set()
self.changed: bool = True
self.prev_access_time = datetime.now()
self.allowed_span = timedelta(seconds=10)
def register(self, callback: Callable[[Any], Any]):
self.callbacks.add(callback)
def unregister(self, callback: Callable[[Any], Any]):
self.callbacks.discard(callback)
def unregister_all(self):
self.callbacks = set()
def access_check(self):
current_time = datetime.now()
if current_time - self.prev_access_time > self.allowed_span:
self.changed = True
self.prev_access_time = current_time
def poll_for_change(self):
self.access_check()
if self.changed:
self.update_all()
self.changed = False
def update_all(self):
for callback in self.callbacks:
callback(self)
def main():
observed = Subject()
observer1 = ConcreteObserver()
observed.register(lambda x: observer1.update(x))
for i in range(10):
print(f'通知済み at {datetime.now()}')
observed.poll_for_change()
time.sleep(3)
if __name__ == '__main__':
main()
通知済み at 2022-06-28 06:50:42.104712
通知あり at 2022-06-28 06:50:42.104712
====================
通知済み at 2022-06-28 06:50:45.122494
通知済み at 2022-06-28 06:50:48.123007
通知済み at 2022-06-28 06:50:51.130357
通知済み at 2022-06-28 06:50:54.131416
通知あり at 2022-06-28 06:50:54.131416
====================
通知済み at 2022-06-28 06:50:57.156377
通知済み at 2022-06-28 06:51:00.159268
通知済み at 2022-06-28 06:51:03.173298
通知済み at 2022-06-28 06:51:06.186092
通知あり at 2022-06-28 06:51:06.186092
====================
通知済み at 2022-06-28 06:51:09.200443
オブザーバーパターンを採用することで解決できる問題は、あるオブジェクトの状態変化に対応したい複数の他のオブジェクトのグループが存在し、かつ、システム内においてそれぞれの結合度合をより「密」にすることなくそれを実現したい場合です。こういった意味では、イベントの管理であったり、ある種のオブジェクトネットワークにおける状態変化への応答、などがターゲットとなるでしょう。
このシリーズでは散々「結合 (coupling)」という言葉を使ってきていますので、ここでもう少し詳しく話しておきたいと思います。「オブジェクト間の結合の度合い」を話題にする場合、あるオブジェクトが知っている必要がある、関係する他のオブジェクトの内部機能・構造etcの情報の量、を指していることが一般的です。「オブジェクト間の結合が疎 (loosely coupled)」であればあるほどお互いに知っている必要がある情報量が少なくて済む、ということであり、結果的により自由度の高いオブジェクト指向システムである、ということができます。結合が疎なシステムにはオブジェクト間の相互依存がほとんど存在せず、それ故に更新も維持もより容易に行うことができます。オブジェクト間の結合度合いを下げることができれば、コード内のある一部に施した変更がコード内の他の部分に意図せず影響を与えてしまう、といったリスクを大幅に軽減することが可能となります。またオブジェクト同士がお互いにほとんど影響を与えていませんから、ユニットテストを実行するのも、トラブルに対処するもの容易です。
さて、オブザーバーパターンの利用が効果的な他の状況としては、従来の Model-View-Controller デザインパターン (このシリーズでも後程取り上げます) や、参照している基のデータが変更されるたびに更新する必要がある出力テキストを表示している場合、などが考えられるでしょう。
基本的に、ある１つのオブジェクト (Subject/Observable) と複数のオブジェクト (Observers) 間に publish-subscribe (出版 - 購読) の関係がある場合はオブザーバーパターンの出番、と考えてよいでしょう。こういったタイプの良い例としては、ネット上でよく見かける newsfeeds、Atom、RSS、podcasts といったフィード (feeds) があります。出版側は購読者側の機能に一切依存していません。完全に「疎」な結合です。オブザーバーパターンにより産み出されるこの「完全な疎」の結合状態によって、プログラム実行時における購読者の追加や削除を非常に簡単に実行することができるようになっています。
今まで見てきた pythonic なオブザーバーパターンの実装をこの章の始めで取り上げた「タスク監視プログラム」に適用してみましょう。新規クラスの追加、既存クラスの更新、変更といった操作がいかに容易であるかを確認してください:
class ObserverBase:
def complete_task(self, observed: 'Task'):
pass
class Wallet(ObserverBase):
def __init__(self):
self.amount: int = 0
def increase_balance(self, amount: int):
self.amount += amount
def decrease_balance(self, amount: int):
self.amount -= amount
def complete_task(self, observed: 'Task'):
self.increase_balance(5)
def __str__(self) -> str:
return str(self.amount)
class Badge(ObserverBase):
def __init__(self, name: str, _type: int):
self.points: int = 0
self.name = name
self._type = _type
self.awarded: bool = False
def add_points(self, amount: int):
self.points += amount
if self.points > 3:
self.awarded = True
def complete_task(self, observed: 'Task'):
if observed._type == self._type:
self.add_points(2)
def __str__(self) -> str:
if self.awarded:
award_string = "所有"
else:
award_string = "----"
return f'{self.name:>13}: {award_string} [{self.points}]'
class User(ObserverBase):
def __init__(self, wallet: Wallet):
self.wallet = wallet
self.badges: list[Badge] = []
self.experience: int = 0
def add_experience(self, amount: int):
self.experience += amount
def complete_task(self, observed: 'Task'):
self.add_experience(1)
def __str__(self) -> str:
NL: str = '\n'
MAX_LEN: int = len(max(['Wallet', 'Experience'], key=lambda x: len(x)))
return (f'{"Wallet":>{MAX_LEN}}:{str(self.wallet):>3}\n'
f'{"Experience":>{MAX_LEN}}:{self.experience:>3}\n'
f'\n++ Badges ++\n{NL.join([str(x) for x in self.badges])}'
f'\n+++++++++++++++++++++++++++++')
class Task:
def __init__(self, user: User, _type: int):
self.observers: set[ObserverBase] = set()
self.user = user
self._type = _type
def register(self, observer: ObserverBase):
self.observers.add(observer)
def unregister(self, observer: ObserverBase):
self.observers.discard(observer)
def complete(self):
for observer in self.observers:
observer.complete_task(self)
def main():
wallet = Wallet()
user = User(wallet)
badges = [
Badge('Fun Badge', 1),
Badge('Bravery Badge', 2),
Badge('Missing Badge', 3)
]
user.badges.extend(badges)
tasks = [Task(user, 1), Task(user, 2), Task(user, 3), Task(user, 2)]
for task in tasks:
task.register(wallet)
task.register(user)
for badge in badges:
task.register(badge)
for task in tasks:
task.complete()
print(user)
if __name__ == "__main__":
main()
ここまでしっかりと自分でコードを打ち込んで実行してきた方であればオブザーバーパターンの理解は完璧なはずです (時間を取ってこの記事をここまで読んでくださった方であっても、コードを自分で打ち込むことはせずコピー＆ペーストで済ませ実行してしまった方は、残念ながら「分かったつもり」であってもその理解は多分「一時的なやり遂げた感」でしかありません。是非自分の手でコードを打ち込み、実行し、デバッグし、機能を追加してください)。このパターンの実装によってシステムが如何に堅固となり、メンテナンスはしやすく、後々の変更や拡張の際にも「嫌々ながら手を付ける」必要がなくなっていることに気が付くはずです。
RSSフィードを購読
