Behavioral patterns

Table of contents

Behavioral design patterns focus on the interaction and communication between objects. They aim to define how objects collaborate, ensuring efficient and flexible communication without tightly coupling components.

Observer

Observer allows a subject (publisher) to maintain a list of its dependents (observers) and automatically notify them of any state changes, usually by calling one of their methods. This pattern is typically used when changing one object requires changing others, and we want to avoid tight coupling between the objects.

Event-based observer

An event-based observer notifies subscribers when a specific event occurs, allowing dynamic reactions without directly monitoring a property change.


from abc import ABC, abstractmethod

class SubscriberInterface(ABC):
    @abstractmethod
    def receive_email(self, newsletter):
        pass

class Event(list): # inheriting from list to store subscribed observers
    def __call__(self, *args, **kwargs): # making the Event instance callable to trigger notifications
        for item in self: # looping through all subscribed observer functions
            item(*args, **kwargs) # calling each observer with the given arguments

class Subscriber(SubscriberInterface):
    def __init__(self, name):
        self.name = name

    def receive_email(self, newsletter):
        print(f"{self.name} received: {newsletter}")

class Newsletter:
    def __init__(self):
        self.new_issue_released = Event()

    def release_new_issue(self, issue_title):
        print(f"Releasing new issue: {issue_title}")
        self.new_issue_released(issue_title)

john = Subscriber("John")
newsletter = Newsletter()
newsletter.new_issue_released.append(john.receive_email) # adding John to the subscription list

newsletter.release_new_issue("Design Patterns Vol. 1") # releasing a new issue
newsletter.new_issue_released.remove(john.receive_email) # removing John from the subscription list
newsletter.release_new_issue("Design Patterns Vol. 2") # releasing a new issue
                                    

Property observer

A property observer ties event handling to a property change (using @property and @setter). It is just a different approach to the Observer pattern. Both are good, and we just have to choose the one that suits our needs.


class Event(list):
    def __call__(self, *args, **kwargs):
        for item in self:
            item(*args, **kwargs)

class Subscriber:
    def __init__(self, name):
        self.name = name

    def receive_email(self, newsletter):
        print(f"{self.name} received: {newsletter}")

class Newsletter:
    def __init__(self):
        self._new_issue = None
        self.new_issue_released = Event() # the event fires when issue updates

    @property
    def new_issue(self):
        return self._new_issue

    @new_issue.setter
    def new_issue(self, issue_title):
        if self._new_issue != issue_title: # ensuring it's actually changing
            self._new_issue = issue_title
            print(f"New issue set: {issue_title}")
            self.new_issue_released(issue_title) # notifying subscribers

john = Subscriber("John")
newsletter = Newsletter()
newsletter.new_issue_released.append(john.receive_email)

newsletter.new_issue = "Design Patterns Vol. 1" # automatically notifying John
newsletter.new_issue_released.remove(john.receive_email)
newsletter.new_issue = "Design Patterns Vol. 2" # no notification sent
                                    

Property dependencies

Property dependencies ensure that one property automatically updates when another changes.


class Person:
    def __init__(self, name, birth_year, current_year):
        self.name = name
        self.birth_year = birth_year
        self.current_year = current_year

    @property
    def age(self):
        return self.current_year - self.birth_year # age depends on birth_year and current_year

p = Person("Bob", 2000, 2025)
print(p.age)
p.current_year = 2030
print(p.age) # age has been automatically updated
                                    

Chain of Responsibility

Chain of Responsibility allows multiple handlers to process a request without the sender needing to know which handler will handle it. The request is passed along a chain of handlers until one of them processes it (or it reaches the end of the chain). This pattern promotes loose coupling and makes it easy to add or modify handlers dynamically. All componenets of a chain get a chance to process a command or a query, and have the ability to terminate the processing chain.

Method chain (classic approach)

In this approach, each handler explicitly calls the next one. It is good for request approval flows.


class Request:
    def __init__(self, amount):
        self.amount = amount

class Handler:
    def __init__(self, successor = None):
        self.successor = successor

    def handle(self, request):
        if self.successor:
            self.successor.handle(request)

class Manager(Handler):
    def handle(self, request):
        if request.amount <= 1000:
            print(f"Manager approves ${request.amount}")
        else:
            super().handle(request)

class Director(Handler):
    def handle(self, request):
        if request.amount <= 5000:
            print(f"Director approves ${request.amount}")
        else:
            super().handle(request)

class CEO(Handler):
    def handle(self, request):
        if request.amount > 5000:
            print(f"CEO approves ${request.amount}")

# Creating the chain: Manager → Director → CEO
manager = Manager(Director(CEO()))

# Requests go through the chain until handled
manager.handle(Request(500)) # handled by Manager
manager.handle(Request(3000)) # handled by Director
manager.handle(Request(10000)) # handled by CEO
                                    

Broker chain (event-based approach)

Broker chain uses an event-driven system where multiple handlers process a request independently without forming a strict chain. It is useful in scenarios where multiple components modify a value dynamically, such as game stat modifiers or discount systems in e-commerce. Even though handlers are not explicitly linked, they act on the request in sequence when the event is triggered. It is a more flexible and decentralized version of the Chain of Responsibility pattern.


class Event(list):
    def __call__(self, *args, **kwargs):
        for item in self:
            item(*args, **kwargs)

class Query:
    def __init__(self, product_name, query_type, value):
        self.product_name = product_name
        self.query_type = query_type
        self.value = value

class ECommercePlatform:
    def __init__(self):
        self.discount_handlers = Event()

    def apply_discounts(self, query):
        self.discount_handlers(query) # calling all registered discount handlers

class Product:
    def __init__(self, platform, name, base_price):
        self.platform = platform
        self.name = name
        self.base_price = base_price

    @property
    def price(self):
        query = Query(self.name, "price", self.base_price)
        self.platform.apply_discounts(query) # applying registered discounts
        return query.value

class DiscountModifier:
    def __init__(self, platform, product):
        self.platform = platform
        self.product = product
        self.platform.discount_handlers.append(self.handle) # registering handler

    def handle(self, query):
        pass # to be implemented by subclasses

class BlackFridayDiscount(DiscountModifier):
    def handle(self, query):
        if query.product_name == self.product.name and query.query_type == "price":
            query.value *= 0.8 # 20% discount

class LoyaltyDiscount(DiscountModifier):
    def handle(self, query):
        if query.product_name == self.product.name and query.query_type == "price":
            query.value *= 0.9 # additional 10% discount for loyal customers

platform = ECommercePlatform()
laptop = Product(platform, "Laptop", 1000)

print("Original price:", laptop.price)

black_friday = BlackFridayDiscount(platform, laptop)
print("After Black Friday discount:", laptop.price)

loyalty = LoyaltyDiscount(platform, laptop)
print("After loyalty discount:", laptop.price)
                                    

Command

Command is used to turn a request (an action) into a stand-alone object. This object encapsulates all details of the request, allowing parameterization of objects. It is also useful for implementing the undo functionality.

In the example below, the TransferMoneyCommand class encapsulates the action and its associated details (it takes the BankAccount objects as parameters) and can be executed by TransactionHistory (the invoker) when triggered.


from abc import ABC, abstractmethod

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

    @abstractmethod
    def undo(self):
        pass

class BankAccount: # a receiver class (the object that performs the action when requested)
    def __init__(self, balance = 0):
        self.balance = balance

    def deposit(self, amount):
        self.balance += amount

    def withdraw(self, amount):
        if self.balance >= amount:
            self.balance -= amount
            return True
        return False

    def __str__(self):
        return f"Balance = {self.balance}"

class TransferMoneyCommand(Command): # a concrete Command classes
    def __init__(self, source_account, target_account, amount):
        self.source_account = source_account
        self.target_account = target_account
        self.amount = amount
        self.executed = False

    def execute(self):
        if not self.executed:
            if self.source_account.withdraw(self.amount):
                self.target_account.deposit(self.amount)
                self.executed = True
                print(f"Transferred {self.amount} from source to target.")
            else:
                print("Insufficient funds for transfer.")
        else:
            print("Transfer already executed.")

    def undo(self):
        if self.executed:
            self.target_account.withdraw(self.amount)
            self.source_account.deposit(self.amount)
            print(f"Undo transfer of {self.amount} from target back to source.")
            self.executed = False
        else:
            print("No transfer to undo.")

class TransactionHistory: # an invoker class (the object that calls the command)
    def __init__(self):
        self.history = []

    def execute_command(self, command):
        command.execute() # invoking the Command's execute method
        self.history.append(command)

    def undo(self):
        if self.history:
            last_command = self.history.pop()
            last_command.undo()

source = BankAccount(500)
target = BankAccount(200)
history = TransactionHistory()

print(f"Source account before transfer: {source}")
print(f"Target account before transfer: {target}")

transfer1 = TransferMoneyCommand(source, target, 100)
history.execute_command(transfer1)

print(f"Source account after first transfer: {source}")
print(f"Target account after first transfer: {target}")

transfer2 = TransferMoneyCommand(source, target, 1000) # trying to transfer more than available funds
history.execute_command(transfer2)

print(f"Source account after second transfer attempt: {source}")
print(f"Target account after second transfer attempt: {target}")
history.undo()

print(f"Source account after undo: {source}")
print(f"Target account after undo: {target}")
                                    

Interpreter

Interpreter is used for designing a system that needs to process and evaluate language expressions. It provides a way to define the grammar of a language and interpret statements in that language. It is commonly used in mathematical expression evaluators, rule-based engines, and parsers. It turns text data into separate lexical tokens (lexing) and interprets sequences of these tokens (parsing).

The example below implements a simple interpreter for arithmetic expressions that includes addition and subtraction. It tokenizes an expression like "10 + 5 - 3", parses it into an abstract syntax tree (AST), and evaluates the result.


from abc import ABC, abstractmethod

# Lexing
class Token:
    NUMBER, PLUS, MINUS = "NUMBER", "PLUS", "MINUS"

    def __init__(self, type_, value = None):
        self.type = type_
        self.value = value

    def __repr__(self):
        return f"Token({self.type}, {self.value})"

class Lexer:
    def __init__(self, text):
        self.text = text.replace(" ", "") # removing spaces for easier processing
        self.index = 0

    def get_next_token(self):
        if self.index >= len(self.text):
            return None

        char = self.text[self.index]
        if char.isdigit():
            num = 0
            # Extracting multi-digit numbers
            while self.index < len(self.text) and self.text[self.index].isdigit():
                num = num * 10 + int(self.text[self.index])
                self.index += 1
            return Token(Token.NUMBER, num)
        elif char == "+":
            self.index += 1
            return Token(Token.PLUS)
        elif char == "-":
            self.index += 1
            return Token(Token.MINUS)

    def tokenize(self):
        tokens = []
        while self.index < len(self.text):
            tokens.append(self.get_next_token()) # generating a token list
        return tokens

# Parsing
class Parser:
    def __init__(self, tokens):
        self.tokens = tokens
        self.current_token = self.tokens.pop(0) if self.tokens else None # fetching the first token

    def eat(self, token_type):
        if self.current_token and self.current_token.type == token_type:
            self.current_token = self.tokens.pop(0) if self.tokens else None # moving to the next token

    def parse(self):
        result = self.current_token.value # starting with the first number
        self.eat(Token.NUMBER)

        while self.current_token:
            if self.current_token.type == Token.PLUS:
                self.eat(Token.PLUS)
                result += self.current_token.value # adding the next number to result
                self.eat(Token.NUMBER)
            elif self.current_token.type == Token.MINUS:
                self.eat(Token.MINUS)
                result -= self.current_token.value # subtracting the next number from result
                self.eat(Token.NUMBER)

        return result # directly returning the result

expression = "10 + 5 - 3"
lexer = Lexer(expression)
tokens = lexer.tokenize()
print(tokens)
parser = Parser(tokens)
number = parser.parse()
print(number)
                                    

Iterator

Iterator provides a way to access elements of a collection sequentially without exposing its underlying structure. Instead of manually managing iteration, the pattern encapsulates it within an iterator object that keeps track of the current position and provides a standard way to retrieve the next element. You can find a basic example of this pattern featuring a custom iterable in the lesson about generators. The example below shows how to use list-backed properties while implementing the Iterator pattern to allow iteration over stored data.


class Inventory:
    def __init__(self, items = None):
        self._items = items if items else []

    @property
    def items(self):
        return self._items

    @items.setter
    def items(self, new_items):
        if isinstance(new_items, list):
            self._items = new_items
        else:
            raise ValueError("Items must be a list.")

    def __iter__(self):
        return iter(self._items) # returning an iterator for the list

inventory = Inventory(["Sword", "Shield", "Potion"])
for item in inventory:
    print(item)

inventory.items = ["Bow", "Arrow", "Helmet"]
for item in inventory:
    print(item)
                                    

Mediator

Mediator defines an object that encapsulates how a set of objects interact. Instead of having multiple objects communicate directly, they communicate through a mediator, which handles the communication and coordination between them. This reduces the dependencies between objects, making the system easier to maintain and scale (the objects do not have to be aware of the other objects).


class Mediator:
    def send(self, message, user):
        pass

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

    def send_message(self, message):
        print(f"{self.name} sends: {message}")
        self.mediator.send(message, self)

    def receive_message(self, message):
        print(f"{self.name} receives: {message}")

class ChatRoom(Mediator):
    def __init__(self):
        self.users = []

    def add_user(self, user):
        self.users.append(user)
        user.mediator = self # setting the chat room as the mediator for each user when they join it

    def send(self, message, user):
        for u in self.users:
            if u != user:
                u.receive_message(message)

chat_room = ChatRoom()

alice = User("Alice")
bob = User("Bob")
charlie = User("Charlie")

chat_room.add_user(alice)
chat_room.add_user(bob)
chat_room.add_user(charlie)

alice.send_message("Hi everyone!")
bob.send_message("Hello Alice!")
charlie.send_message("Hey all!")                                    
                                    

Memento

Memento allows capturing and restoring an object's state without exposing its internal details. It is useful for implementing undo/redo functionality or saving checkpoints in an application.


class Memento: # storing the state of the text document
    def __init__(self, content):
        self.content = content

class TextEditor:
    def __init__(self):
        self.content = ""
        self.history = [Memento(self.content)]
        self.current_state = 0

    def write(self, text):
        self.content += text
        m = Memento(self.content)
        self.history.append(m)
        self.current_state += 1

    def undo(self):
        if self.current_state > 0:
            self.current_state -= 1
            m = self.history[self.current_state]
            self.content = m.content
            return m
        return None

    def redo(self):
        if self.current_state + 1 < len(self.history):
            self.current_state += 1
            m = self.history[self.current_state]
            self.content = m.content
            return m
        return None

    def show_content(self):
        print(f"Document content: {self.content}")

editor = TextEditor()
editor.write("Hello, ")
editor.show_content()
editor.write("world!")
editor.show_content()

editor.undo()
editor.show_content()
editor.redo()
editor.show_content()
                                    

State

State allows an object to change its behavior when its internal state changes. It encapsulates different behaviors within separate state classes, making the system more maintainable and flexible.

In the example below, the behavior that changes is how the TrafficLight object responds to the change() method, depending on its current state. This method changes the state, and depending on whether the current state is RedState, YellowState, or GreenState, a different transition logic contained in the switch() method of each state class is applied.


from abc import ABC, abstractmethod

class TrafficLightState(ABC): # a State interface
    @abstractmethod
    def switch(self, light):
        pass

# Concrete State classes
class RedState(TrafficLightState):
    def switch(self, light):
        print("Red -> Switching to Green")
        light.state = GreenState()

class YellowState(TrafficLightState):
    def switch(self, light):
        print("Yellow -> Switching to Red")
        light.state = RedState()

class GreenState(TrafficLightState):
    def switch(self, light):
        print("Green -> Switching to Yellow")
        light.state = YellowState()

class TrafficLight: # a context class (the object that maintains the current state and triggers state transitions)
    def __init__(self):
        self.state = RedState() # an initial State

    def change(self):
        self.state.switch(self) # switching the State by executing the switch() method from the current State class

traffic_light = TrafficLight()
traffic_light.change()
traffic_light.change()
traffic_light.change()
                                    

Strategy

Strategy defines a family of algorithms, encapsulates each one in a separate class, and allows them to be interchangeable without modifying the client code. This enables selecting an algorithm dynamically at runtime, promoting flexibility and maintainability.


from abc import ABC, abstractmethod

class SortStrategy(ABC): # a Strategy interface
    @abstractmethod
    def sort(self, data):
        pass

# Concrete Strategy classes
class QuickSort(SortStrategy):
    def sort(self, data):
        print("Sorting using QuickSort")
        return sorted(data) # a placeholder for actual implementation

class MergeSort(SortStrategy):
    def sort(self, data):
        print("Sorting using MergeSort")
        return sorted(data) # a placeholder for actual implementation
    
class BubbleSort(SortStrategy):
    def sort(self, data):
        print("Sorting using BubbleSort")
        return sorted(data) # a placeholder for actual implementation

class Sorter: # a context class
    def __init__(self, strategy: SortStrategy):
        self.strategy = strategy

    def set_strategy(self, strategy: SortStrategy):
        self.strategy = strategy # changing Strategies dynamically

    def sort_list(self, data):
        return self.strategy.sort(data)

numbers = [5, 2, 9, 1, 5, 6]

sorter = Sorter(QuickSort())
print(sorter.sort_list(numbers))

sorter.set_strategy(MergeSort())
print(sorter.sort_list(numbers))

sorter.set_strategy(BubbleSort())
print(sorter.sort_list(numbers))
                                    

Template Method

Template Method defines the skeleton of an algorithm in a base class while allowing subclasses to fill in specific steps. This promotes code reuse and ensures that the core logic remains unchanged while allowing flexibility in certain steps.


from abc import ABC, abstractmethod

class ReportTemplate(ABC):
    def generate_report(self): # a Template Method defining the steps for generating a report
        self.collect_data()
        self.analyze_data()
        print("Done.")
        
    @abstractmethod
    def collect_data(self):
        pass

    @abstractmethod
    def analyze_data(self):
        pass

class SalesReport(ReportTemplate):
    def collect_data(self):
        print("Collecting sales data...")

    def analyze_data(self):
        print("Analyzing sales trends...")

class InventoryReport(ReportTemplate):
    def collect_data(self):
        print("Collecting inventory data...")

    def analyze_data(self):
        print("Checking stock levels...")

sales = SalesReport()
sales.generate_report()

inventory = InventoryReport()
inventory.generate_report()
                                    

Visitor

Visitor allows us to add new operations to existing object structures without modifying them (in compliance with the OCP). Instead of adding behavior inside the already existing objects, we create a Visitor that implements the behavior and visits those objects. This keeps the document structure unchanged while allowing new operations.


from abc import ABC, abstractmethod

class DocumentElement(ABC): # an element interface
    @abstractmethod
    def accept(self, visitor):
        pass

# Concrete elements
class Text(DocumentElement):
    def accept(self, visitor):
        visitor.visit_text(self)

class Image(DocumentElement):
    def accept(self, visitor):
        visitor.visit_image(self)

class Visitor(ABC): # a Visitor interface
    @abstractmethod
    def visit_text(self, text):
        pass

    @abstractmethod
    def visit_image(self, image):
        pass

# Concrete Visitors
class Renderer(Visitor):
    def visit_text(self, text):
        print("Rendering text")

    def visit_image(self, image):
        print("Rendering image")

class WordCounter(Visitor):
    def visit_text(self, text):
        print("Counting words in text")

    def visit_image(self, image):
        print("Skipping word count for image")

elements = [Text(), Image()]
renderer = Renderer()
word_counter = WordCounter()
for element in elements:
    element.accept(renderer)
    element.accept(word_counter)
                                    

Instead of adding new methods to Text and Image every time we need a new operation, we simply create a new Visitor class that implements those operations without modifying the element classes. This pattern relies on double dispatch, where both the element type (Text or Image) and the visitor type (Renderer or WordCounter) determine the behavior, meaning:

  1. The element calls its accept(visitor) method, passing the visitor as an argument.
  2. Inside accept(visitor), the element calls the visitor’s corresponding method (visit_text() or visit_image()), passing itself (self) as an argument.
  3. The visitor determines what to do based on the element type.
Visitor VS Decorator

A Visitor allows new operations to be added to existing object structures without modifying them by separating the new behavior from the objects it operates on. Decorator allows behavior to be dynamically added to an object at runtime by wrapping it with additional functionality and extending behavior.