Skip to content

Behavioural Design Patterns#

Start here

Read the Design Patterns overview. Strategy, Observer, and Iterator map directly to common interview questions — see the overview table.

Behavioral patterns focus on how objects communicate, how responsibilities are distributed, and how algorithms or workflows can vary at runtime. These patterns make complex interaction between objects more flexible and easier to extend.

Memento Pattern#

Purpose

Capture and externalize an object's internal state so it can be restored later, without exposing implementation details. A Caretaker stores Mementos; an Originator creates and restores them.

When to use

  • You need undo/redo or snapshot/rollback (editors, forms, games).
  • You want to preserve encapsulation — the caretaker should not inspect or modify the saved state.
  • State history must be managed separately from the object that owns the state.
memento_graphic_editor.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from dataclasses import dataclass
from typing import List


@dataclass(frozen=True)
class EditorMemento:
    """Immutable snapshot of a shape's attributes."""

    shape_type: str
    x: int
    y: int
    color: str
    size: int


class GraphicEditor:
    """Originator: edits shapes and creates/restores mementos."""

    def __init__(self) -> None:
        self.shape_type = ""
        self.x = 0
        self.y = 0
        self.color = ""
        self.size = 0

    def set_shape(self, shape_type: str, x: int, y: int, color: str, size: int) -> None:
        self.shape_type = shape_type
        self.x = x
        self.y = y
        self.color = color
        self.size = size

    def save(self) -> EditorMemento:
        return EditorMemento(self.shape_type, self.x, self.y, self.color, self.size)

    def restore(self, memento: EditorMemento) -> None:
        self.shape_type = memento.shape_type
        self.x = memento.x
        self.y = memento.y
        self.color = memento.color
        self.size = memento.size

    def get_shape(self) -> str:
        return (
            f"Shape: {self.shape_type}, Position: ({self.x}, {self.y}), "
            f"Color: {self.color}, Size: {self.size}"
        )


class Caretaker:
    """Stores memento history and restores previous states."""

    def __init__(self) -> None:
        self.history: List[EditorMemento] = []

    def save_state(self, editor: GraphicEditor) -> None:
        self.history.append(editor.save())

    def undo(self, editor: GraphicEditor) -> None:
        if not self.history:
            return
        self.history.pop()
        if self.history:
            editor.restore(self.history[-1])


def demo() -> None:
    editor = GraphicEditor()
    caretaker = Caretaker()

    shapes = [
        ("circle", 10, 20, "red", 5),
        ("square", 30, 40, "blue", 10),
        ("triangle", 50, 60, "green", 15),
    ]
    for shape in shapes:
        editor.set_shape(*shape)
        caretaker.save_state(editor)

    caretaker.undo(editor)
    print(editor.get_shape())


if __name__ == "__main__":
    demo()

Explanation#

  1. EditorMemento is immutable (frozen=True) so saved snapshots cannot be altered after creation.
  2. GraphicEditor is the originator — it knows how to serialize and deserialize its own state.
  3. Caretaker only stores and retrieves mementos; it never reads individual fields inside a snapshot.
  4. Each save_state call pushes a snapshot onto the history stack before the next edit.
  5. undo pops the latest snapshot and restores the previous one, enabling rollback without breaking encapsulation.

Interview note

Memento often appears alongside Command (undo stacks) and in system-design discussions about audit trails or version history. Mention that mementos can grow memory — consider limiting history depth in production.

Observer Pattern#

Purpose

Define a one-to-many dependency so when one object (the Subject) changes state, all dependent Observers are notified automatically.

When to use

  • Multiple components must react to the same event (stock ticks, UI updates, logging).
  • You want loose coupling — subjects should not know concrete observer classes.
  • You are building pub/sub, event buses, or reactive pipelines.
observer_stock_market.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from abc import ABC, abstractmethod
from typing import List


class Observer(ABC):
    @abstractmethod
    def update(self, stock_symbol: str, new_price: float) -> None:
        pass


class Subject(ABC):
    @abstractmethod
    def register_observer(self, observer: Observer) -> None:
        pass

    @abstractmethod
    def remove_observer(self, observer: Observer) -> None:
        pass

    @abstractmethod
    def notify_observers(self, stock_symbol: str, new_price: float) -> None:
        pass


class StockMarket(Subject):
    def __init__(self, price_change_threshold: float) -> None:
        self._observers: List[Observer] = []
        self.price_change_threshold = price_change_threshold

    def register_observer(self, observer: Observer) -> None:
        self._observers.append(observer)

    def remove_observer(self, observer: Observer) -> None:
        self._observers.remove(observer)

    def notify_observers(self, stock_symbol: str, new_price: float) -> None:
        for observer in self._observers:
            observer.update(stock_symbol, new_price)

    def set_stock_price(self, stock_symbol: str, new_price: float, old_price: float) -> None:
        price_change = abs(new_price - old_price) / old_price * 100
        if price_change >= self.price_change_threshold:
            self.notify_observers(stock_symbol, new_price)


class InvestorA(Observer):
    def update(self, stock_symbol: str, new_price: float) -> None:
        print(f"Investor A notified: Stock {stock_symbol} has a new price: ${new_price}")


class InvestorB(Observer):
    def update(self, stock_symbol: str, new_price: float) -> None:
        print(f"Investor B notified: Stock {stock_symbol} has a new price: ${new_price}")


def demo() -> None:
    market = StockMarket(price_change_threshold=5.0)
    investor_a = InvestorA()
    investor_b = InvestorB()

    market.register_observer(investor_a)
    market.register_observer(investor_b)

    updates = [
        ("AAPL", 150.0, 140.0),
        ("GOOG", 2800.0, 2750.0),
        ("MSFT", 310.0, 300.0),
        ("TSLA", 700.0, 650.0),
        ("AMZN", 3300.0, 3200.0),
    ]

    for i, (symbol, new_price, old_price) in enumerate(updates, start=1):
        if i == 5:
            market.remove_observer(investor_b)
        market.set_stock_price(symbol, new_price, old_price)


if __name__ == "__main__":
    demo()

Explanation#

  1. Observer defines the callback contract (update) that all subscribers implement.
  2. Subject manages a dynamic list of observers and broadcasts changes through notify_observers.
  3. StockMarket only notifies when price movement exceeds the configured threshold.
  4. InvestorA and InvestorB are concrete observers with independent reaction logic.
  5. remove_observer on the fifth update demonstrates dynamic unsubscription without restarting the subject.

Interview note

Observer maps to event-driven design, WebSocket feeds, and reactive UIs. In interviews, contrast it with Mediator (central coordinator) and Pub/Sub (message broker between anonymous publishers/subscribers).

Strategy Pattern#

Purpose

Define a family of interchangeable algorithms, encapsulate each one, and make them swappable at runtime without changing the client code.

When to use

  • You have multiple ways to perform the same task (sorting, formatting, pricing, routing).
  • Conditional if/elif chains for algorithm selection are growing unwieldy.
  • You want to open/closed behavior — add new strategies without editing the context class.
strategy_document_formatter.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from abc import ABC, abstractmethod


class TextFormatter(ABC):
    @abstractmethod
    def format(self, text: str) -> str:
        pass


class PlainTextFormatter(TextFormatter):
    def format(self, text: str) -> str:
        return text


class HTMLFormatter(TextFormatter):
    def format(self, text: str) -> str:
        return f"<html><body>{text}</body></html>"


class MarkdownFormatter(TextFormatter):
    def format(self, text: str) -> str:
        return f"**{text}**"


class Document:
    def __init__(self) -> None:
        self.content = ""
        self.formatter: TextFormatter = PlainTextFormatter()

    def set_content(self, content: str) -> None:
        self.content = content

    def set_formatter(self, formatter: TextFormatter) -> None:
        self.formatter = formatter

    def display(self) -> None:
        print(self.formatter.format(self.content))


def demo() -> None:
    document = Document()
    document.set_content("Design patterns simplify complex software design.")

    document.set_formatter(PlainTextFormatter())
    print("Plain Text:")
    document.display()

    document.set_formatter(HTMLFormatter())
    print("HTML Format:")
    document.display()

    document.set_formatter(MarkdownFormatter())
    print("Markdown Format:")
    document.display()


if __name__ == "__main__":
    demo()

Explanation#

  1. TextFormatter is the strategy interface — one method, many implementations.
  2. PlainTextFormatter, HTMLFormatter, and MarkdownFormatter encapsulate formatting rules.
  3. Document is the context — it holds content and delegates formatting to the active strategy.
  4. set_formatter swaps behavior at runtime without modifying display.
  5. New output formats can be added by implementing TextFormatter; Document stays unchanged.

Interview note

Strategy is a frequent FAANG pattern for "how would you support multiple pricing/shipping/sorting rules?" Python's first-class functions can replace explicit strategy classes for simple cases, but the class-based form is clearer in OOP interviews.

Command Pattern#

Purpose

Encapsulate a request as an object, letting you parameterize clients, queue operations, log them, and support undo.

When to use

  • Actions must be scheduled, retried, logged, or undone (remote controls, job queues, macros).
  • Invokers should not depend on receiver implementation details.
  • You want to compose or batch operations declaratively.
command_remote_control.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
from abc import ABC, abstractmethod
from typing import Optional


class Command(ABC):
    @abstractmethod
    def execute(self) -> None:
        pass


class Light:
    def turn_on(self) -> None:
        print("The light is on.")

    def turn_off(self) -> None:
        print("The light is off.")


class Fan:
    def turn_on(self) -> None:
        print("The fan is on.")

    def turn_off(self) -> None:
        print("The fan is off.")


class LightOnCommand(Command):
    def __init__(self, light: Light) -> None:
        self.light = light

    def execute(self) -> None:
        self.light.turn_on()


class LightOffCommand(Command):
    def __init__(self, light: Light) -> None:
        self.light = light

    def execute(self) -> None:
        self.light.turn_off()


class FanOnCommand(Command):
    def __init__(self, fan: Fan) -> None:
        self.fan = fan

    def execute(self) -> None:
        self.fan.turn_on()


class FanOffCommand(Command):
    def __init__(self, fan: Fan) -> None:
        self.fan = fan

    def execute(self) -> None:
        self.fan.turn_off()


class RemoteControl:
    def __init__(self) -> None:
        self.light_on_command: Optional[Command] = None
        self.light_off_command: Optional[Command] = None
        self.fan_on_command: Optional[Command] = None
        self.fan_off_command: Optional[Command] = None

    def set_light_on_command(self, command: Command) -> None:
        self.light_on_command = command

    def set_light_off_command(self, command: Command) -> None:
        self.light_off_command = command

    def set_fan_on_command(self, command: Command) -> None:
        self.fan_on_command = command

    def set_fan_off_command(self, command: Command) -> None:
        self.fan_off_command = command

    def press_light_on_button(self) -> None:
        if self.light_on_command:
            self.light_on_command.execute()

    def press_light_off_button(self) -> None:
        if self.light_off_command:
            self.light_off_command.execute()

    def press_fan_on_button(self) -> None:
        if self.fan_on_command:
            self.fan_on_command.execute()

    def press_fan_off_button(self) -> None:
        if self.fan_off_command:
            self.fan_off_command.execute()


def demo() -> None:
    light = Light()
    fan = Fan()

    remote = RemoteControl()
    remote.set_light_on_command(LightOnCommand(light))
    remote.set_light_off_command(LightOffCommand(light))
    remote.set_fan_on_command(FanOnCommand(fan))
    remote.set_fan_off_command(FanOffCommand(fan))

    remote.press_light_on_button()
    remote.press_light_off_button()
    remote.press_fan_on_button()
    remote.press_fan_off_button()


if __name__ == "__main__":
    demo()

Explanation#

  1. Command turns an action into an object with a uniform execute method.
  2. Light and Fan are receivers — they perform the actual work.
  3. Concrete commands (LightOnCommand, etc.) bind a receiver to a specific action.
  4. RemoteControl is the invoker — it triggers commands without knowing receiver internals.
  5. Commands can be stored, replayed, or extended with undo() using Memento for state snapshots.

Interview note

Command appears in task queues, transaction logs, and undo stacks. Pair it with Memento when interviewers ask how undo would work after executing a command.

Template Method Pattern#

Purpose

Define the skeleton of an algorithm in a base class, letting subclasses override specific steps without changing the overall structure.

When to use

  • Several classes share the same workflow but differ in a few steps (ETL pipelines, report generation, data import).
  • You want to enforce a fixed sequence while allowing customization hooks.
  • Subclasses should not be able to reorder critical steps.
template_report_generator.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from abc import ABC, abstractmethod


class ReportTemplate(ABC):
    def generate_report(self) -> None:
        """Template method — defines the fixed algorithm skeleton."""
        self.gather_data()
        self.process_data()
        self.format_report()
        self.print_report()

    @abstractmethod
    def gather_data(self) -> None:
        pass

    @abstractmethod
    def process_data(self) -> None:
        pass

    def format_report(self) -> None:
        print("Formatting the report with appropriate layout and style.")

    def print_report(self) -> None:
        print("Printing the report for final review and distribution.")


class SalesReport(ReportTemplate):
    def gather_data(self) -> None:
        print("Gathering sales figures from regional databases.")

    def process_data(self) -> None:
        print("Aggregating revenue, units sold, and growth rates.")


class EmployeeReport(ReportTemplate):
    def gather_data(self) -> None:
        print("Collecting employee attendance and performance records.")

    def process_data(self) -> None:
        print("Computing averages, rankings, and department summaries.")


class InventoryReport(ReportTemplate):
    def gather_data(self) -> None:
        print("Fetching stock levels from warehouse systems.")

    def process_data(self) -> None:
        print("Identifying low-stock items and reorder thresholds.")


def demo() -> None:
    reports = [SalesReport(), EmployeeReport(), InventoryReport()]
    for report in reports:
        print(f"Generating {report.__class__.__name__}:")
        report.generate_report()
        print()


if __name__ == "__main__":
    demo()

Explanation#

  1. ReportTemplate.generate_report is the template method — it owns the step order.
  2. gather_data and process_data are abstract hooks subclasses must implement.
  3. format_report and print_report provide shared default behavior all reports reuse.
  4. Each concrete report (SalesReport, EmployeeReport, InventoryReport) customizes only the variable steps.
  5. Subclasses cannot skip format_report or reorder steps without overriding the entire template.

Interview note

Template Method is the inheritance-based cousin of Strategy (composition). In interviews, explain when you'd prefer hooks in a base class vs injecting a strategy object for flexibility.

Iterator Pattern#

Purpose

Provide a way to traverse elements of a collection without exposing its internal representation.

When to use

  • Clients should iterate over composite structures (trees, graphs, custom collections) uniformly.
  • You want to hide whether the backing store is a list, queue, set, or tree.
  • You need multiple traversal orders (in-order, level-order, filtered views).
iterator_notifications.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
from abc import ABC, abstractmethod
from collections import deque
from typing import Deque, Iterator, List, Set


class Notification:
    def __init__(self, message: str) -> None:
        self.message = message

    def get_message(self) -> str:
        return self.message


class NotificationCollection(ABC):
    @abstractmethod
    def create_iterator(self) -> Iterator[Notification]:
        pass


class EmailNotification(NotificationCollection):
    def __init__(self) -> None:
        self._notifications: List[Notification] = []

    def add_notification(self, message: str) -> None:
        self._notifications.append(Notification(message))

    def create_iterator(self) -> Iterator[Notification]:
        return iter(self._notifications)


class SMSNotification(NotificationCollection):
    def __init__(self) -> None:
        self._notifications: Deque[Notification] = deque()

    def add_notification(self, message: str) -> None:
        self._notifications.append(Notification(message))

    def create_iterator(self) -> Iterator[Notification]:
        return iter(list(self._notifications))


class PushNotification(NotificationCollection):
    def __init__(self) -> None:
        self._notifications: Set[Notification] = set()
        self._order: List[Notification] = []

    def add_notification(self, message: str) -> None:
        notification = Notification(message)
        if notification not in self._notifications:
            self._notifications.add(notification)
            self._order.append(notification)

    def create_iterator(self) -> Iterator[Notification]:
        return iter(self._order)


class NotificationManager:
    def __init__(self) -> None:
        self.email_notifications = EmailNotification()
        self.sms_notifications = SMSNotification()
        self.push_notifications = PushNotification()

    def add_email_notification(self, message: str) -> None:
        self.email_notifications.add_notification(message)

    def add_sms_notification(self, message: str) -> None:
        self.sms_notifications.add_notification(message)

    def add_push_notification(self, message: str) -> None:
        self.push_notifications.add_notification(message)

    def print_all_notifications(self) -> None:
        self._print_notifications(self.email_notifications.create_iterator(), "Email")
        self._print_notifications(self.sms_notifications.create_iterator(), "SMS")
        self._print_notifications(self.push_notifications.create_iterator(), "Push")

    def _print_notifications(self, iterator: Iterator[Notification], channel: str) -> None:
        print(f"{channel} Notifications:")
        for notification in iterator:
            print(notification.get_message())


def demo() -> None:
    manager = NotificationManager()
    batches = [
        ("Welcome email", "OTP via SMS", "App update available"),
        ("Invoice attached", "Delivery confirmed", "Flash sale now live"),
    ]
    for email, sms, push in batches:
        manager.add_email_notification(email)
        manager.add_sms_notification(sms)
        manager.add_push_notification(push)

    manager.print_all_notifications()


if __name__ == "__main__":
    demo()

Explanation#

  1. NotificationCollection defines a factory method create_iterator for uniform traversal.
  2. EmailNotification backs storage with a list; iteration follows insertion order.
  3. SMSNotification uses a deque but still exposes a standard Python iterator to callers.
  4. PushNotification uses a set for uniqueness while preserving insertion order for iteration.
  5. NotificationManager prints all channels without knowing each collection's internal structure.

Interview note

Iterator is heavily tested in FAANG interviews. Practice the real design problem: LeetCode 173 — BST Iterator (Tree § BST Iterator). For custom data-structure design (LRU order traversal, stream iterators), see Challenge Yourself.

State Pattern#

Purpose

Let an object alter its behavior when its internal state changes, as if the object changed its class.

When to use

  • Object behavior depends on state and must switch at runtime (media players, TCP connections, order workflows).
  • Large if/elif state checks make the class hard to maintain.
  • Each state should encapsulate its own transition rules.
state_media_player.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
from abc import ABC, abstractmethod


class State(ABC):
    @abstractmethod
    def press_play(self) -> None:
        pass

    @abstractmethod
    def press_stop(self) -> None:
        pass

    @abstractmethod
    def press_pause(self) -> None:
        pass

    @abstractmethod
    def display(self) -> None:
        pass


class PlayingState(State):
    def press_play(self) -> None:
        print("Starting playback")

    def press_stop(self) -> None:
        print("Stopping playback")

    def press_pause(self) -> None:
        print("Pausing playback")

    def display(self) -> None:
        print("Current State: Playing")


class PausedState(State):
    def press_play(self) -> None:
        print("Resuming playback")

    def press_stop(self) -> None:
        print("Stopping playback from pause")

    def press_pause(self) -> None:
        print("Pausing playback")

    def display(self) -> None:
        print("Current State: Paused")


class StoppedState(State):
    def press_play(self) -> None:
        print("Starting playback")

    def press_stop(self) -> None:
        print("Stopping playback")

    def press_pause(self) -> None:
        print("Can't pause. Media is already stopped")

    def display(self) -> None:
        print("Current State: Stopped")


class MediaPlayer:
    def __init__(self) -> None:
        self._state: State = PlayingState()

    def set_state(self, state: State) -> None:
        self._state = state

    def play(self) -> None:
        self._state.press_play()

    def stop(self) -> None:
        self._state.press_stop()

    def pause(self) -> None:
        self._state.press_pause()

    def display_state(self) -> None:
        self._state.display()


def demo(command: str = "Pause") -> None:
    player = MediaPlayer()

    if command == "Play":
        player.play()
    elif command == "Pause":
        player.set_state(PausedState())
        player.pause()
    elif command == "Stop":
        player.set_state(StoppedState())
        player.stop()
    else:
        print("Invalid choice.")
        return

    player.display_state()


if __name__ == "__main__":
    demo()

Explanation#

  1. State defines the interface for all behavior variants (play, stop, pause, display).
  2. PlayingState, PausedState, and StoppedState each implement context-specific responses.
  3. MediaPlayer delegates every action to the current state object instead of branching on enums.
  4. set_state swaps behavior at runtime — the player object itself stays the same.
  5. Invalid transitions (e.g., pause while stopped) are handled inside the relevant state class.

Interview note

State vs Strategy: State objects often drive transitions themselves; Strategy objects are usually chosen externally and do not change the context's mode. State machines are common in workflow and connection-lifecycle questions.

Mediator Pattern#

Purpose

Define an object that encapsulates how a set of objects interact, promoting loose coupling by preventing peers from referring to each other directly.

When to use

  • Many objects communicate in complex ways (chat rooms, air-traffic control, UI dialog coordination).
  • Direct references between components would create a dense dependency web.
  • Centralized coordination rules (resource limits, routing, permissions) are easier to maintain in one place.
mediator_flight_control.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from abc import ABC, abstractmethod
from typing import List


class Mediator(ABC):
    @abstractmethod
    def register_airplane(self, airplane: "Airplane") -> None:
        pass

    @abstractmethod
    def handle_takeoff_request(self, airplane: "Airplane") -> None:
        pass

    @abstractmethod
    def handle_landing_request(self, airplane: "Airplane") -> None:
        pass


class Airplane:
    def __init__(self, airplane_id: str) -> None:
        self.id = airplane_id
        self._mediator: Mediator | None = None

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

    def request_takeoff(self) -> None:
        print(f"Airplane {self.id} requesting takeoff")
        if self._mediator:
            self._mediator.handle_takeoff_request(self)

    def request_landing(self) -> None:
        print(f"Airplane {self.id} requesting landing")
        if self._mediator:
            self._mediator.handle_landing_request(self)

    def receive_notification(self, message: str) -> None:
        print(f"Airplane {self.id}: {message}")


class ControlTower(Mediator):
    def __init__(self, takeoff_runways: int = 2, landing_runways: int = 2) -> None:
        self.airplanes: List[Airplane] = []
        self.takeoff_runways = takeoff_runways
        self.landing_runways = landing_runways

    def register_airplane(self, airplane: Airplane) -> None:
        self.airplanes.append(airplane)
        airplane.set_mediator(self)

    def handle_takeoff_request(self, airplane: Airplane) -> None:
        if self.takeoff_runways > 0:
            self.takeoff_runways -= 1
            self._notify_airplane(
                airplane, f"Takeoff approved. Runways available: {self.takeoff_runways}"
            )
        else:
            self._notify_airplane(
                airplane, "Takeoff denied. No runways available. Please wait"
            )

    def handle_landing_request(self, airplane: Airplane) -> None:
        if self.landing_runways > 0:
            self.landing_runways -= 1
            self._notify_airplane(
                airplane, f"Landing approved. Runways available: {self.landing_runways}"
            )
        else:
            self._notify_airplane(
                airplane, "Landing denied. No runways available. Please wait"
            )

    def complete_takeoff(self, airplane: Airplane) -> None:
        print(f"Airplane {airplane.id} has taken off")
        self.takeoff_runways += 1
        print(f"Runway freed. Available takeoff runways: {self.takeoff_runways}")

    def complete_landing(self, airplane: Airplane) -> None:
        print(f"Airplane {airplane.id} has landed")
        self.landing_runways += 1
        print(f"Runway freed. Available landing runways: {self.landing_runways}")

    def _notify_airplane(self, airplane: Airplane, message: str) -> None:
        airplane.receive_notification(message)


def demo() -> None:
    control_tower = ControlTower()
    airplanes = [Airplane(f"PL-{i}") for i in range(1, 5)]

    for airplane in airplanes:
        control_tower.register_airplane(airplane)

    airplanes[0].request_takeoff()
    airplanes[1].request_takeoff()
    airplanes[2].request_takeoff()
    airplanes[3].request_takeoff()

    control_tower.complete_takeoff(airplanes[0])
    control_tower.complete_takeoff(airplanes[1])

    airplanes[2].request_takeoff()
    airplanes[3].request_takeoff()

    control_tower.complete_takeoff(airplanes[2])
    control_tower.complete_takeoff(airplanes[3])

    airplanes[0].request_landing()
    airplanes[1].request_landing()
    control_tower.complete_landing(airplanes[0])
    control_tower.complete_landing(airplanes[1])

    airplanes[2].request_landing()
    airplanes[3].request_landing()
    control_tower.complete_landing(airplanes[2])
    control_tower.complete_landing(airplanes[3])


if __name__ == "__main__":
    demo()

Explanation#

  1. Mediator defines the coordination contract for registering and handling requests.
  2. Airplane colleagues send requests to the mediator instead of talking to each other.
  3. ControlTower centralizes runway allocation and approval/denial logic.
  4. complete_takeoff and complete_landing free resources and update availability counters.
  5. Adding a new colleague (e.g., helicopter) only requires registering with the mediator — no mesh of peer references.

Interview note

Mediator centralizes coordination; Observer broadcasts events. In system design, mediators resemble orchestrators or API gateways that enforce policies so individual services stay decoupled.