Python メディエーターパターン

メディエーターパターンは、1つの中間者（メディエーター）オブジェクトを導入することで、オブジェクト間の直接的な通信依存を減らす、振る舞いに関するデザインパターンです。現実の空港管制塔を想像してください。すべての航空機が直接通信するのではなく、管制塔を通じて離着陸や航路を調整することで、混乱や衝突を避けています。

ソフトウェア開発では、複数のオブジェクト間に複雑な相互作用がある場合、メディエーターパターンはそれらの相互作用をメディエーター内にカプセル化することで、オブジェクト間の結合度を下げることができます。

中核となる考え方

メディエーターパターンの中核となる考え方は「直接話さず、メディエーターを通じて調整する」。これにより、オブジェクト間の多対多の相互作用の複雑さが解決され、システムの保守と拡張が容易になります。

なぜメディエーターパターンが必要か

問題シナリオ

チャットルームシステムを開発しているとします。複数のユーザーが互いにメッセージを送信する必要があります。メディエーターがない場合、コードは次のようになるでしょう：

例

class User:
def __init__(self, name):
self.name = name
self.other_users = [] # 他のすべてのユーザーを知る必要がある

def send_message(self, message, to_user):
print(f"{self.name} が {to_user.name} に送信: {message}")
to_user.receive_message(message, self)

def receive_message(self, message, from_user):
print(f"{self.name} が {from_user.name} からのメッセージを受信: {message}")

# 使用例
user1 = User("小明")
user2 = User("小红")
user3 = User("小刚")

# 各ユーザーは他のすべてのユーザーを知る必要がある
user1.other_users = [user2, user3]
user2.other_users = [user1, user3]
user3.other_users = [user1, user2]

user1.send_message("こんにちは！", user2)

この設計の欠点：

オブジェクト間の結合度が高く、各ユーザーが他のすべてのユーザーを知る必要がある
新しいユーザーを追加する際に既存のすべてのユーザーを変更する必要がある
保守が難しく、相互作用のロジックが各オブジェクトに分散している

解決策

メディエーターパターンは、チャットルーム（メディエーター）を導入することでこの問題を解決します：

例

class ChatRoom:
def __init__(self):
self.users = {}

def register_user(self, user):
self.users[user.name] = user
user.chat_room = self

def send_message(self, message, from_user, to_user_name=None):
if to_user_name: # 個別チャット
if to_user_name in self.users:
self.users[to_user_name].receive_message(message, from_user)
else: # グループチャット
for user_name, user in self.users.items():
if user_name != from_user.name:
user.receive_message(message, from_user)

class User:
def __init__(self, name):
self.name = name
self.chat_room = None

def send_message(self, message, to_user_name=None):
if self.chat_room:
self.chat_room.send_message(message, self, to_user_name)

def receive_message(self, message, from_user):
print(f"{self.name} が {from_user.name} からのメッセージを受信: {message}")

# 使用例
chat_room = ChatRoom()

user1 = User("小明")
user2 = User("小红")
user3 = User("小刚")

chat_room.register_user(user1)
chat_room.register_user(user2)
chat_room.register_user(user3)

user1.send_message("みなさん、こんにちは！") # 一斉送信メッセージ
user2.send_message("こんにちは、小明！", "小明") # 個別チャット

メディエーターパターンの実装

基本構造

より完全な例を通して、メディエーターパターンの実装を理解しましょう：

例

from abc import ABC, abstractmethod
from typing import List

# 抽象メディエーター
class Mediator(ABC):
@abstractmethod
def notify(self, sender: object, event: str, data: dict = None):
pass

# 具体的メディエーター - チャットルーム
class ChatRoomMediator(Mediator):
def __init__(self):
self.users: List[User] = []

def add_user(self, user):
self.users.append(user)
user.set_mediator(self)

def notify(self, sender, event, data=None):
if event == "send_message":
message = data.get("message")
target_user = data.get("target_user")

if target_user: # 個別チャット
for user in self.users:
if user.name == target_user:
user.receive_message(message, sender.name)
else: # グループチャット
for user in self.users:
if user != sender:
user.receive_message(message, sender.name)

elif event == "user_joined":
message = f"システム: {sender.name} がチャットルームに参加しました"
for user in self.users:
if user != sender:
user.receive_message(message, "システム")

# 基本コンポーネントクラス
class BaseComponent:
def __init__(self):
self._mediator = None

def set_mediator(self, mediator: Mediator):
self._mediator = mediator

# 具体的コンポーネント - ユーザー
class User(BaseComponent):
def __init__(self, name):
super().__init__()
self.name = name

def send_message(self, message, target_user=None):
print(f"{self.name} がメッセージを送信: {message}")
self._mediator.notify(self, "send_message", {
"message": message,
"target_user": target_user
})

def join_chat(self):
self._mediator.notify(self, "user_joined")

def receive_message(self, message, from_name):
print(f"{self.name} が {from_name} からのメッセージを受信: {message}")

# 使用例
def main():
# メディエーターを作成
chat_room = ChatRoomMediator()

# ユーザーを作成
alice = User("Alice")
bob = User("Bob")
charlie = User("Charlie")

# チャットルームにユーザーを登録
chat_room.add_user(alice)
chat_room.add_user(bob)
chat_room.add_user(charlie)

print("=== チャットルームデモ ===")

# Alice がチャットルームに参加
alice.join_chat()

# メッセージを送信
alice.send_message("みなさん、こんにちは、Aliceです！")
bob.send_message("Alice、ようこそ！")
charlie.send_message("Alice、こんにちは！", "Alice") # 個別チャット

# Bob がグループメッセージを送信
bob.send_message("誰かオンラインですか？")

if __name__ == "__main__":
main()

実行結果

=== 聊天室演示 ===
系统: Alice 加入了聊天室
Alice 发送消息: 大家好，我是 Alice！
Bob 收到来自 Alice 的消息: 大家好，我是 Alice！
Charlie 收到来自 Alice 的消息: 大家好，我是 Alice！
Bob 发送消息: 欢迎 Alice！
Alice 收到来自 Bob 的消息: 欢迎 Alice！
Charlie 收到来自 Bob 的消息: 欢迎 Alice！
Charlie 发送消息: Alice 你好！
Alice 收到来自 Charlie 的消息: Alice 你好！
Bob 发送消息: 有人在线吗？
Alice 收到来自 Bob 的消息: 有人在线吗？
Charlie 收到来自 Bob 的消息: 有人在线吗？

メディエーターパターンのUML構造

構造の説明

Mediator（メディエーターインターフェース）：通信インターフェースを定義する
ConcreteMediator（具体的メディエーター）：調整ロジックを実装し、すべてのコンポーネントを把握する
BaseComponent（基本コンポーネント）：メディエーターへの参照を含む
ConcreteComponent（具体的コンポーネント）：具体的なビジネスロジックを実装する

実際の適用シナリオ

シナリオ1：GUIアプリケーション

グラフィカルユーザーインターフェースでは、さまざまなコントロール（ボタン、テキストボックス、チェックボックスなど）がメディエーターを通じて相互作用を調整します：

例

class DialogMediator:
def __init__(self):
self.login_button = None
self.username_input = None
self.password_input = None
self.remember_checkbox = None

def notify(self, sender, event):
if event == "username_changed" or event == "password_changed":
# ユーザー名またはパスワードが入力されたとき、ログインボタンを有効にできるか確認する
username = self.username_input.get_text()
password = self.password_input.get_text()
self.login_button.set_enabled(bool(username and password))

elif event == "login_clicked":
# ログインロジックを処理する
username = self.username_input.get_text()
password = self.password_input.get_text()
remember = self.remember_checkbox.is_checked()
print(f"ログイン: {username}, ログイン状態を保持: {remember}")

elif event == "remember_changed":
print("ログイン状態を保持のオプションが変更されました")

class UIComponent:
def __init__(self, mediator=None):
self.mediator = mediator

def set_mediator(self, mediator):
self.mediator = mediator

class Button(UIComponent):
def click(self):
if self.mediator:
self.mediator.notify(self, "login_clicked")

def set_enabled(self, enabled):
print(f"ボタン {'有効' if enabled else '無効'}")

class TextInput(UIComponent):
def __init__(self, mediator=None):
super().__init__(mediator)
self._text = ""

def set_text(self, text):
self._text = text
if self.mediator:
self.mediator.notify(self, "username_changed")

def get_text(self):
return self._text

# 使用例
dialog = DialogMediator()

login_btn = Button()
username_input = TextInput()
password_input = TextInput()

dialog.login_button = login_btn
dialog.username_input = username_input
dialog.password_input = password_input

login_btn.set_mediator(dialog)
username_input.set_mediator(dialog)
password_input.set_mediator(dialog)

# ユーザー入力をシミュレート
username_input.set_text("user123")
password_input.set_text("pass123")
login_btn.click()

シナリオ2：EC注文システム

ECシステムでは、注文処理が在庫、決済、物流など複数のサブシステムに関係します：

例

class OrderMediator:
def __init__(self):
self.inventory_system = None
self.payment_system = None
self.shipping_system = None
self.notification_system = None

def place_order(self, order_data):
print("=== 注文処理を開始 ===")

# 在庫を確認
if not self.inventory_system.check_stock(order_data):
return "在庫不足"

# 決済を処理
payment_result = self.payment_system.process_payment(order_data)
if not payment_result:
return "決済失敗"

# 在庫を更新
self.inventory_system.update_stock(order_data)

# 出荷を手配
shipping_info = self.shipping_system.schedule_delivery(order_data)

# 通知を送信
self.notification_system.send_confirmation(order_data, shipping_info)

return "注文処理が成功しました"

class InventorySystem:
def check_stock(self, order_data):
print(に庫を確認中...)
return True

def update_stock(self, order_data):
print(に庫を更新中...)

class PaymentSystem:
def process_payment(self, order_data):
print("決済を処理中...")
return True

class ShippingSystem:
def schedule_delivery(self, order_data):
print("出荷を手配中...")
return "配送伝票番号: SF123456789"

class NotificationSystem:
def send_confirmation(self, order_data, shipping_info):
print(f"確認メールを送信、出荷情報: {shipping_info}")

# 使用例
mediator = OrderMediator()

mediator.inventory_system = InventorySystem()
mediator.payment_system = PaymentSystem()
mediator.shipping_system = ShippingSystem()
mediator.notification_system = NotificationSystem()

order_data = {"product_id": "123", "quantity": 2, "user_id": "user001"}
result = mediator.place_order(order_data)
print(f"注文結果: {result}")

メディエーターパターンの長所と短所

長所

長所	説明
単一責任の原則	コンポーネント間の相互作用ロジックをメディエーターに集中させる
開放/閉鎖の原則（オープン・クローズド原則）	既存のコンポーネントを変更せずに新しいメディエーターを導入できる
結合度を下げる	コンポーネント間の直接的な依存を減らす
再利用が容易	個々のコンポーネントを異なるコンテキストでより容易に再利用できる

短所

短所	説明
メディエーターが複雑になる可能性がある	相互作用ロジックが増えるにつれて、メディエーターが神オブジェクトになる可能性がある
パフォーマンスのオーバーヘッド	すべての通信がメディエーターを経由するため、パフォーマンスのボトルネックが生じる可能性がある
デバッグが困難	相互作用ロジックがメディエーターに集中しているため、デバッグ時に複数のコンポーネントを追跡する必要があるかもしれない

ベストプラクティスと注意点

1. メディエーターの責務を適切に分割する

1つのメディエーターにすべてを処理させず、業務領域に応じて複数のメディエーターに分割することができます：

例

class UserRegistrationMediator:
def notify(self, sender, event, data):
if event == "user_registered":
# ユーザー登録後の関連ロジックを処理
self.send_welcome_email(data)
self.create_user_profile(data)
self.add_to_mailing_list(data)

class OrderProcessingMediator:
def notify(self, sender, event, data):
if event == "order_created":
# 注文作成後の関連ロジックを処理
self.validate_order(data)
self.process_payment(data)
self.update_inventory(data)

2. メディエーターの過度な複雑化を避ける

メディエーターが複雑になりすぎた場合は、他のパターンの使用やリファクタリングを検討してください：

例

# 良くない例 - メディエーターにロジックが多すぎる
class ComplexMediator:
def handle_event(self, event_type, data):
if event_type == "user_action":
# 大量の複雑なビジネスロジック...
pass
elif event_type == "system_event":
# さらに複雑なロジック...
pass

# 良い方法 - ストラテジーパターンを使用して複雑なロジックを分解する
class EventHandler:
def handle(self, data):
pass

class UserActionHandler(EventHandler):
def handle(self, data):
# ユーザー操作を専門に処理する
pass

class SimpleMediator:
def __init__(self):
self.handlers = {}

def register_handler(self, event_type, handler):
self.handlers[event_type] = handler

def notify(self, event_type, data):
if event_type in self.handlers:
self.handlers[event_type].handle(data)

3. オブザーバーパターンとの組み合わせを検討する

メディエーターパターンはしばしばオブザーバーパターンと組み合わせて使用されます：

例

from abc import ABC, abstractmethod
from typing import List

class Observer(ABC):
@abstractmethod
def update(self, event, data):
pass

class Observable:
def __init__(self):
self._observers: List[Observer] = []

def add_observer(self, observer):
self._observers.append(observer)

def notify_observers(self, event, data=None):
for observer in self._observers:
observer.update(event, data)

class ChatMediator(Observable):
def send_message(self, message, sender, receiver=None):
# メディエーターのロジック
event_data = {
"message": message,
"sender": sender,
"receiver": receiver
}
self.notify_observers("message_sent", event_data)

class LoggingObserver(Observer):
def update(self, event, data):
if event == "message_sent":
print(f"ログ: {data['sender']} がメッセージを送信しました")

練習問題

練習1：チャットルームシステムの改良

以前のチャットルームシステムを拡張し、以下の機能を追加します：

複数のチャットルームの作成をサポート
ユーザーが異なるチャットルーム間を切り替えられるようにする
管理者機能の追加（キック、ミュートなど）

練習2：航空券予約システムの設計

メディエーターパターンを使用して航空券予約システムを設計し、以下のコンポーネントを含めます：

フライト検索
座席選択
支払い処理
チケット発行
通知送信

各コンポーネントがメディエーターを通じて協調して動作することを要件とします。

練習3：スマートホーム制御システム

スマートホーム制御システムを設計し、以下を含めます：

照明制御
温度調節
セキュリティ監視
シーンモード

メディエーターを通じてデバイス間のスマートな連携を実現します。

まとめ

メディエーターパターンは、複雑なオブジェクト間の相互作用を処理するための強力なツールです。調整センターを導入することでシステムアーキテクチャを簡素化し、特に以下のシナリオに適しています：

オブジェクト間に複雑な網状の相互作用関係が存在する場合
複数のオブジェクト間の相互作用ロジックを集中的に制御する必要がある場合
オブジェクト間の結合度を低減したい場合

覚えておいてください、メディエーターパターンは万能ではありません。相互作用ロジックが非常に単純な場合は、オブジェクト間の直接通信の方が適切かもしれません。しかし、複雑なシステムを扱う場合、メディエーターパターンを適切に使用することで、コードの保守性と拡張性を大幅に向上させることができます。

重要なポイント：

メディエーターはオブジェクト間の相互作用ロジックをカプセル化する
コンポーネントはメディエーターとのみ通信し、直接相互依存しない
メディエーターは複雑になる可能性があるため、適切な設計が必要
他のパターン（オブザーバーなど）と組み合わせることで、より良い効果が得られる

その他の拡張