Mediator
The Mediator defines an object that encapsulates how a set of objects interact in order to reduce direct dependencies between them. It promotes loose coupling by centralizing communication and restricting direct communication - objects must always collaborate only via the mediator
TL;DR
The Mediator pattern reduces direct dependencies between objects by having them communicate through a central mediator.
Problem
Suppose you are creating a chat room with multiple users in it. Whenever a user sends a message, every other user should be able to see the message. A naive solution would be to make every user aware of every other user, but this leads to very tightly coupled code that becomes very difficult to maintain.
Solution
A solution is to use a Mediator ChatRoom object that is aware of all “subscribers” or people in the room. This way, each user is simply aware of the Mediator and uses it as an interface to send all messages. The Mediator itself is responsible for broadcasting the information to any other interested party.
The key point is that users should not be aware at all about the existence of other users.
Another example of this is aircraft requesting landing and performing other events in relation to an air traffic control tower. Each plane does not need to know anything about other planes, runways, or emergency services - they simply communicate with ATC, and ATC manages the communications with the rest of the system. Note how the events are all driven by the aircraft itself - it reports landings, emergencies, etc - and everything else ocurrs as a result of the notification to ATC.
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
# --- Event System using @dataclass ---
@dataclass
class Event:
sender: object
@dataclass
class RequestLanding(Event):
pass
@dataclass
class LandingComplete(Event):
pass
@dataclass
class EmergencyLanding(Event):
message: str
# --- Mediator Interface ---
class Mediator(ABC):
@abstractmethod
def notify(self, event: Event) -> None:
pass
# --- Concrete Mediator ---
class AirTrafficControl(Mediator):
def __init__(self, plane: Plane, runway: Runway, emergency_team: EmergencyTeam) -> None:
self.plane = plane
self.runway = runway
self.emergency_team = emergency_team
self.plane.mediator = self
self.runway.mediator = self
self.emergency_team.mediator = self
def notify(self, event: Event) -> None:
match event:
case RequestLanding():
print("AirTrafficControl: Plane requested landing.")
self.runway.prepare()
case LandingComplete():
print("AirTrafficControl: Landing complete.")
self.emergency_team.stand_down()
case EmergencyLanding(message=msg):
print(f"AirTrafficControl: Emergency reported! Message: {msg}")
self.runway.clear()
self.emergency_team.respond()
# --- Base Component ---
class BaseComponent:
def __init__(self, mediator: Mediator = None) -> None:
self._mediator = mediator
@property
def mediator(self) -> Mediator:
return self._mediator
@mediator.setter
def mediator(self, mediator: Mediator) -> None:
self._mediator = mediator
# --- Components ---
class Plane(BaseComponent):
def request_landing(self) -> None:
print("Plane: Requesting landing.")
self.mediator.notify(RequestLanding(self))
def emergency_landing(self) -> None:
print("Plane: Declaring emergency!")
self.mediator.notify(EmergencyLanding(self, message="Engine failure"))
def landed(self) -> None:
self.mediator.notify(LandingComplete(self))
class Runway(BaseComponent):
def prepare(self) -> None:
print("Runway: Cleared and ready for landing.")
def clear(self) -> None:
print("Runway: Emergency! Clearing runway immediately!")
class EmergencyTeam(BaseComponent):
def respond(self) -> None:
print("EmergencyTeam: Responding to emergency.")
def stand_down(self) -> None:
print("EmergencyTeam: Standing down. No action needed.")
# --- Client Code ---
if __name__ == "__main__":
plane = Plane()
runway = Runway()
emergency_team = EmergencyTeam()
atc = AirTrafficControl(plane, runway, emergency_team)
print("✈️ Normal landing sequence:")
plane.request_landing()
plane.landed()
print("\n⚠️ Emergency landing sequence:")
plane.emergency_landing()
A more complex example
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List
# --- Events using @dataclass ---
@dataclass
class Event:
sender: object
@dataclass
class RequestLanding(Event):
pass
@dataclass
class LandingComplete(Event):
pass
@dataclass
class EmergencyLanding(Event):
message: str
# --- Mediator Interface ---
class Mediator(ABC):
@abstractmethod
def notify(self, event: Event) -> None:
pass
# --- Components Base ---
class BaseComponent:
def __init__(self, mediator: Mediator = None):
self._mediator = mediator
@property
def mediator(self) -> Mediator:
return self._mediator
@mediator.setter
def mediator(self, mediator: Mediator):
self._mediator = mediator
# --- Components ---
class Plane(BaseComponent):
def __init__(self, name: str):
super().__init__()
self.name = name
def request_landing(self):
print(f"🛬 {self.name}: Requesting landing.")
self.mediator.notify(RequestLanding(self))
def emergency_landing(self, message="Unknown emergency"):
print(f"🚨 {self.name}: Declaring emergency!")
self.mediator.notify(EmergencyLanding(self, message))
def landed(self):
print(f"✅ {self.name}: Landed successfully.")
self.mediator.notify(LandingComplete(self))
class Runway(BaseComponent):
def prepare_for(self, plane: Plane):
print(f"🛩 Runway: Cleared for landing: {plane.name}")
def clear(self):
print("⚠️ Runway: Emergency! Clearing runway!")
class EmergencyTeam(BaseComponent):
def respond(self, plane: Plane):
print(f"🚑 EmergencyTeam: Responding to {plane.name}'s emergency!")
def stand_down(self):
print("✅ EmergencyTeam: Standing down. No further action needed.")
# --- Concrete Mediator with Landing Queue ---
class AirTrafficControl(Mediator):
def __init__(self, runway: Runway, emergency_team: EmergencyTeam):
self.runway = runway
self.emergency_team = emergency_team
self.runway.mediator = self
self.emergency_team.mediator = self
self.landing_queue: List[Plane] = []
self.current_plane: Plane | None = None
def register_plane(self, plane: Plane):
plane.mediator = self
def notify(self, event: Event):
match event:
case RequestLanding(sender=plane):
if self.current_plane is None:
self.current_plane = plane
print(f"ATC: {plane.name} is cleared to land.")
self.runway.prepare_for(plane)
else:
print(f"ATC: {plane.name} added to queue.")
self.landing_queue.append(plane)
case LandingComplete(sender=plane):
if plane == self.current_plane:
self.emergency_team.stand_down()
self.current_plane = None
if self.landing_queue:
next_plane = self.landing_queue.pop(0)
self.current_plane = next_plane
print(f"ATC: {next_plane.name} is now cleared to land.")
self.runway.prepare_for(next_plane)
case EmergencyLanding(sender=plane, message=msg):
print(f"ATC: Emergency from {plane.name}: {msg}")
self.runway.clear()
self.emergency_team.respond(plane)
# Emergency plane gets immediate landing
if plane in self.landing_queue:
self.landing_queue.remove(plane)
if self.current_plane:
self.landing_queue.insert(0, self.current_plane)
self.current_plane = plane
self.runway.prepare_for(plane)
# --- Simulation ---
if __name__ == "__main__":
runway = Runway()
emergency_team = EmergencyTeam()
atc = AirTrafficControl(runway, emergency_team)
# Register multiple planes
plane1 = Plane("Flight A123")
plane2 = Plane("Flight B456")
plane3 = Plane("Flight C789")
for plane in (plane1, plane2, plane3):
atc.register_plane(plane)
print("\n--- Normal Landing Requests ---")
plane1.request_landing()
plane2.request_landing()
plane3.request_landing()
print("\n--- First Plane Lands ---")
plane1.landed()
print("\n--- Emergency! ---")
plane3.emergency_landing("Hydraulic failure")
print("\n--- Emergency Plane Lands ---")
plane3.landed()
print("\n--- Remaining Plane Lands ---")
plane2.landed()