Design Pattern

2025-09-14 9940 words 47 minutes

Contents

Creational Design Patterns:

What are Creational Design Patterns?

Creational patterns focus on how objects are created, making the system independent of how its objects are created, composed, and represented. They solve problems that arise when basic object creation could result in design problems or added complexity due to inflexible creation procedures. geeksforgeeks The five main creational patterns are:

Singleton - Ensures only one instance exists
Factory Method - Creates objects through subclasses
Abstract Factory - Creates families of related objects
Builder - Constructs complex objects step-by-step
Prototype - Creates objects by copying existing instances

Singleton Pattern

Structure and Implementation

Factory Method Pattern - Payment Processing System Structure

Python Implementation

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class DatabaseConnection:
    """Thread-safe Singleton implementation"""
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super(DatabaseConnection, cls).__new__(cls)
                    cls._instance._initialized = False
        return cls._instance
    
    def __init__(self):
        if not self._initialized:
            self.connection_string = "postgresql://localhost:5432/mydb"
            self.is_connected = False
            self._initialized = True

When to use Singleton

Use Singleton when:

Managing database connection pools
Implementing configuration managers
Creating system-wide logging services
Managing cache systems or thread pools
Controlling hardware interface access

Avoid when:

Multiple instances might be needed in the future
Unit testing becomes difficult due to global state
Multithreading issues aren’t properly handled

Best Practices

Implement thread-safe initialization using double-checked locking
Consider lazy initialization to improve startup performance
Make clone() and copy constructors private to prevent duplication
Document the singleton constraint clearly
Consider dependency injection as an alternative stackoverflow

Factory Method Pattern

The Factory Method pattern provides an interface for creating objects but allows subclasses to determine which concrete class to instantiate. realpython refactoring.guru

Structure and Implementation

Abstract Factory Pattern - Cross-Platform UI Components Structure

Python implementation

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class PaymentProcessor(ABC):
    @abstractmethod
    def create_payment_method(self) -> PaymentMethod:
        pass
    
    def process_transaction(self, amount: float, payment_data: Dict) -> str:
        payment_method = self.create_payment_method()
        if payment_method.validate_payment(payment_data):
            return payment_method.process_payment(amount)
        return "Payment validation failed"

class CreditCardProcessor(PaymentProcessor):
    def create_payment_method(self) -> PaymentMethod:
        return CreditCardPayment()

When to Use Factory Method

Use Factory Method when:

Creating objects without specifying exact classes
Subclasses need to alter the type of objects created
Object creation involves complex logic
Supporting plugin architectures or framework extension points. geeksforgeeks

Avoid when:

Simple object creation is sufficient
Only one type of object will ever be created
Performance is critical (adds abstraction overhead)

Best Practices

Keep factory methods focused solely on object creation
Use meaningful parameter names for clarity
Handle invalid inputs gracefully with appropriate exceptions
Consider using enums for type parameters to improve type safety
Document factory method contracts and supported types clearly

Abstract Factory Pattern

The Abstract Factory pattern creates families of related objects without specifying their concrete classes. It acts as a “factory of factories”. refactoring.guru realpython

Structure and Implementation

Builder Pattern - Computer Configuration System Structure

Python Implementation

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class UIFactory(ABC):
    @abstractmethod
    def create_button(self) -> Button:
        pass
    
    @abstractmethod
    def create_text_field(self) -> TextField:
        pass

class WindowsUIFactory(UIFactory):
    def create_button(self) -> Button:
        return WindowsButton()
    
    def create_text_field(self) -> TextField:
        return WindowsTextField()

class Application:
    def __init__(self, factory: UIFactory):
        self.factory = factory
        self.button = factory.create_button()
        self.text_field = factory.create_text_field()

When to Use Abstract Factory

Use Factory Method when:

Creating objects without specifying exact classes
Subclasses need to alter the type of objects created
Object creation involves complex logic
Supporting plugin architectures or framework extension points. geeksforgeeks

Avoid when:

Simple object creation is sufficient
Only one type of object will ever be created
Performance is critical (adds abstraction overhead)

Best Practices

Keep factory methods focused solely on object creation
Use meaningful parameter names for clarity
Handle invalid inputs gracefully with appropriate exceptions
Consider using enums for type parameters to improve type safety
Document factory method contracts and supported types clearly

Builder Pattern

The Builder pattern constructs complex objects step-by-step, allowing the same construction process to create different representations. refactoring.guru geeksforgeeks

Structure and Implementation

Python Implementation

 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
class ComputerBuilder(ABC):
    def __init__(self):
        self.computer = Computer()
    
    @abstractmethod
    def build_cpu(self): pass
    
    @abstractmethod
    def build_ram(self): pass
    
    def get_computer(self) -> Computer:
        return self.computer

class GamingComputerBuilder(ComputerBuilder):
    def build_cpu(self):
        self.computer.cpu = "Intel Core i9-13900K"
        self.computer.price += 589.99
        return self  # Enable method chaining
    
    def build_ram(self):
        self.computer.ram = "32GB DDR5-6000 RGB"
        self.computer.price += 299.99
        return self

class ComputerDirector:
    def construct_computer(self, builder: ComputerBuilder) -> Computer:
        return (builder.build_cpu()
                      .build_ram()
                      .build_storage()
                      .get_computer())

When to Use Builder

Use Builder when:

Objects have many parameters (especially optional ones)
Objects require step-by-step construction with validation
Creating immutable objects with complex initialization
Different representations of the same object are needed

Avoid when:

Objects are simple with few parameters
Object creation is straightforward
Performance overhead is unacceptable for simple objects

Best Practices

Support method chaining (fluent interface) for better readability
Perform validation in the build() method rather than during construction
Use meaningful method names that clearly indicate what’s being built
Provide sensible default values for optional parameters
Consider making built objects immutable for thread safety

Prototype Pattern

The Prototype pattern creates new objects by copying existing instances, avoiding expensive creation operations. softwarepatterns refactoring.guru

Structure and Implementation

Game Character Cloning System Structure

Python Implementation

 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
class GameCharacter(ABC):
    @abstractmethod
    def clone(self) -> 'GameCharacter':
        pass

class Warrior(GameCharacter):
    def __init__(self, name: str = "Warrior", level: int = 1):
        self.name = name
        self.level = level
        self.skills = ["Sword Strike", "Shield Block"]
        self.equipment = {"weapon": "Iron Sword", "armor": "Leather Armor"}
    
    def clone(self) -> 'Warrior':
        cloned = Warrior(self.name, self.level)
        cloned.skills = self.skills.copy()  # Shallow copy
        cloned.equipment = self.equipment.copy()
        return cloned

# Using Python's built-in copy module
import copy

class AdvancedCharacter:
    def shallow_clone(self):
        return copy.copy(self)
    
    def deep_clone(self):
        return copy.deepcopy(self)

When to Use Prototype

Use Prototype when

Object creation is expensive (database operations, network calls) geeksforgeeks
Objects are similar to existing ones with minor variations
Dynamic object types need to be determined at runtime
Avoiding subclass explosion for object variations

Avoid when

Object creation is cheap and straightforward
Objects don’t have complex state that benefits from copying
Deep copying operations are more expensive than creation
Objects require unique identities that shouldn’t be duplicated

Best Practices

Choose between shallow and deep copying appropriately
Use Python’s copy module when suitable for built-in copying functionality
Implement prototype registry pattern for managing multiple prototypes
Handle circular references carefully in deep copying scenarios
Reset mutable shared state after cloning to avoid unintended side effects

Pattern Selection Guide

Choosing the Right Creational Design Pattern

Real-World Application

Enterprise Application

Database Connection Management: Singleton pattern for connection pools in web applications
Configuration Management: Singleton or Builder for application settings
API Gateway Selection: Factory Method for choosing different payment or cloud service providers

Game Development

Character Creation: Prototype pattern for game character templates with different configurations
Asset Loading: Abstract Factory for platform-specific game assets (graphics, audio, input handlers)
Level Construction: Builder pattern for complex game level creation with multiple components

Web Development

Request Processing: Factory Method for creating different request handlers based on content type
Theme Systems: Abstract Factory for creating UI component families for different themes
Form Builders: Builder pattern for creating complex forms with validation and conditional fields

Structural Design Patterns

Structural Design Patterns are fundamental architectural blueprints that focus on how classes and objects are composed to form larger, more complex structures while maintaining flexibility and efficiency. These patterns serve as the architects of software systems, ensuring components are well-connected, organized, and scalable. structural-design-patterns when-to-use-which-design-pattern

Core Purpose and Principles

Structural patterns solve the critical problem of object assembly by identifying simple ways to realize relationships among entities. They enable developers to compose objects into tree structures, adapt incompatible interfaces, and create simplified facades for complex subsystems . The two fundamental themes driving these patterns are making independently developed class libraries work together and describing ways to compose objects to realize new functionality.

Adapter Pattern

The Adapter Pattern acts as a bridge between two incompatible interfaces, allowing them to work together seamlessly. This pattern is particularly valuable when integrating legacy systems with modern architectures or when working with third-party libraries that don’t match your existing interface. geeksforgeeks

When to Use:

Protocol conversion between different communication standards
Legacy system integration with modern interfaces
Interface standardization across various sensor outputs

Python Implementation

 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
from abc import ABC, abstractmethod

# Target interface that client expects
class MediaPlayer(ABC):
    @abstractmethod
    def play(self, audio_type: str, filename: str):
        pass

# Legacy audio player with incompatible interface
class Mp3Player:
    def play_mp3(self, filename: str):
        print(f"Playing MP3 file: {filename}")

class Mp4Player:
    def play_mp4(self, filename: str):
        print(f"Playing MP4 file: {filename}")

# Adapter that makes legacy players compatible
class AudioAdapter(MediaPlayer):
    def __init__(self):
        self.mp3_player = Mp3Player()
        self.mp4_player = Mp4Player()
    
    def play(self, audio_type: str, filename: str):
        if audio_type.lower() == "mp3":
            self.mp3_player.play_mp3(filename)
        elif audio_type.lower() == "mp4":
            self.mp4_player.play_mp4(filename)
        else:
            print(f"Unsupported audio type: {audio_type}")

# Client code using the adapter
class AudioPlayer(MediaPlayer):
    def __init__(self):
        self.adapter = AudioAdapter()
    
    def play(self, audio_type: str, filename: str):
        if audio_type.lower() == "mp3" or audio_type.lower() == "mp4":
            self.adapter.play(audio_type, filename)
        else:
            print(f"Cannot play {audio_type} format")

# Usage
player = AudioPlayer()
player.play("mp3", "song.mp3")
player.play("mp4", "video.mp4")

Architecture Diagram

Bridge Pattern

The Bridge Pattern decouples an abstraction from its implementation, allowing both to vary independently. This pattern uses composition over inheritance and is designed up-front to let abstractions and implementations evolve separately.

When to Use

When both abstraction and implementation need to change independently
Avoiding permanent binding between abstraction and implementation
Sharing implementation among multiple objects

Python Implementation

 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
from abc import ABC, abstractmethod

# Implementation interface
class RenderAPI(ABC):
    @abstractmethod
    def render_circle(self, x: float, y: float, radius: float):
        pass
    
    @abstractmethod
    def render_rectangle(self, x: float, y: float, width: float, height: float):
        pass

# Concrete implementations
class OpenGLRenderer(RenderAPI):
    def render_circle(self, x: float, y: float, radius: float):
        print(f"OpenGL: Drawing circle at ({x}, {y}) with radius {radius}")
    
    def render_rectangle(self, x: float, y: float, width: float, height: float):
        print(f"OpenGL: Drawing rectangle at ({x}, {y}) with size {width}x{height}")

class DirectXRenderer(RenderAPI):
    def render_circle(self, x: float, y: float, radius: float):
        print(f"DirectX: Rendering circle at ({x}, {y}) with radius {radius}")
    
    def render_rectangle(self, x: float, y: float, width: float, height: float):
        print(f"DirectX: Rendering rectangle at ({x}, {y}) with size {width}x{height}")

# Abstraction
class Shape(ABC):
    def __init__(self, renderer: RenderAPI):
        self.renderer = renderer
    
    @abstractmethod
    def draw(self):
        pass

# Refined abstractions
class Circle(Shape):
    def __init__(self, x: float, y: float, radius: float, renderer: RenderAPI):
        super().__init__(renderer)
        self.x = x
        self.y = y
        self.radius = radius
    
    def draw(self):
        self.renderer.render_circle(self.x, self.y, self.radius)

class Rectangle(Shape):
    def __init__(self, x: float, y: float, width: float, height: float, renderer: RenderAPI):
        super().__init__(renderer)
        self.x = x
        self.y = y
        self.width = width
        self.height = height
    
    def draw(self):
        self.renderer.render_rectangle(self.x, self.y, self.width, self.height)

# Usage
opengl = OpenGLRenderer()
directx = DirectXRenderer()

shapes = [
    Circle(10, 20, 5, opengl),
    Rectangle(30, 40, 100, 50, directx),
    Circle(0, 0, 15, directx)
]

for shape in shapes:
    shape.draw()

Architecture Diagram

Composite Pattern

The Composite Pattern composes objects into tree structures to represent part-whole hierarchies, allowing clients to treat individual objects and compositions uniformly. This pattern is essential for modeling recursive structures where components can contain other components.

When to Use

Graphics systems with shapes and groups of shapes
File systems representing files and directories
UI frameworks handling complex nested components

Python Implementation

 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
from abc import ABC, abstractmethod
from typing import List

# Component interface
class FileSystemComponent(ABC):
    @abstractmethod
    def get_size(self) -> int:
        pass
    
    @abstractmethod
    def display(self, indent: int = 0):
        pass

# Leaf component
class File(FileSystemComponent):
    def __init__(self, name: str, size: int):
        self.name = name
        self.size = size
    
    def get_size(self) -> int:
        return self.size
    
    def display(self, indent: int = 0):
        print("  " * indent + f"📄 {self.name} ({self.size} bytes)")

# Composite component
class Directory(FileSystemComponent):
    def __init__(self, name: str):
        self.name = name
        self.children: List[FileSystemComponent] = []
    
    def add(self, component: FileSystemComponent):
        self.children.append(component)
    
    def remove(self, component: FileSystemComponent):
        self.children.remove(component)
    
    def get_size(self) -> int:
        return sum(child.get_size() for child in self.children)
    
    def display(self, indent: int = 0):
        print("  " * indent + f"📁 {self.name}/ ({self.get_size()} bytes total)")
        for child in self.children:
            child.display(indent + 1)

# Usage - Building a file system structure
root = Directory("root")
documents = Directory("Documents")
images = Directory("Images")

# Add files to directories
documents.add(File("resume.pdf", 1024))
documents.add(File("report.docx", 2048))

images.add(File("photo1.jpg", 5120))
images.add(File("photo2.png", 3072))

# Create nested structure
root.add(documents)
root.add(images)
root.add(File("readme.txt", 512))

# Display the entire structure
root.display()
print(f"\nTotal size: {root.get_size()} bytes")

Architecture Diagram

Decorator Patterns

The Decorator Pattern attaches additional responsibilities to objects dynamically, providing a flexible alternative to subclassing for extending functionality . This pattern allows behavior to be added to objects without altering their structure. geeksforgeeks

When to Use

Feature extension like logging, error handling, or encryption
Signal processing with filtering or amplification
Dynamic component configuration

Python Implementation

 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
from abc import ABC, abstractmethod
from functools import wraps
import time

# Component interface
class Coffee(ABC):
    @abstractmethod
    def cost(self) -> float:
        pass
    
    @abstractmethod
    def description(self) -> str:
        pass

# Concrete component
class SimpleCoffee(Coffee):
    def cost(self) -> float:
        return 2.00
    
    def description(self) -> str:
        return "Simple coffee"

# Base decorator
class CoffeeDecorator(Coffee):
    def __init__(self, coffee: Coffee):
        self._coffee = coffee
    
    def cost(self) -> float:
        return self._coffee.cost()
    
    def description(self) -> str:
        return self._coffee.description()

# Concrete decorators
class MilkDecorator(CoffeeDecorator):
    def cost(self) -> float:
        return self._coffee.cost() + 0.50
    
    def description(self) -> str:
        return self._coffee.description() + ", milk"

class SugarDecorator(CoffeeDecorator):
    def cost(self) -> float:
        return self._coffee.cost() + 0.25
    
    def description(self) -> str:
        return self._coffee.description() + ", sugar"

class WhipDecorator(CoffeeDecorator):
    def cost(self) -> float:
        return self._coffee.cost() + 0.75
    
    def description(self) -> str:
        return self._coffee.description() + ", whipped cream"

# Function decorator example for logging
def log_execution_time(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} executed in {end_time - start_time:.4f} seconds")
        return result
    return wrapper

# Usage
@log_execution_time
def prepare_coffee():
    # Start with simple coffee
    coffee = SimpleCoffee()
    
    # Add decorators
    coffee = MilkDecorator(coffee)
    coffee = SugarDecorator(coffee)
    coffee = WhipDecorator(coffee)
    
    return coffee

# Test the decorator pattern
my_coffee = prepare_coffee()
print(f"Order: {my_coffee.description()}")
print(f"Total cost: ${my_coffee.cost():.2f}")

Architecture Diagram

Facade Pattern

The Facade Pattern provides a simplified interface to a complex subsystem, hiding the complexity from clients. This pattern creates a unified interface that makes the subsystem easier to use while maintaining the full functionality underneath.

When to Use

Microcontroller libraries requiring complex initialization
Communication systems with multiple protocols
Power management with unified control interfaces

Python Implementation

 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
# Complex subsystem classes
class CPU:
    def freeze(self):
        print("CPU: Freezing processor")
    
    def jump(self, position: int):
        print(f"CPU: Jumping to position {position}")
    
    def execute(self):
        print("CPU: Executing instructions")

class Memory:
    def load(self, position: int, data: str):
        print(f"Memory: Loading '{data}' at position {position}")

class HardDrive:
    def read(self, lba: int, size: int) -> str:
        print(f"HardDrive: Reading {size} bytes from LBA {lba}")
        return f"boot_data_{lba}"

class Graphics:
    def initialize(self):
        print("Graphics: Initializing graphics subsystem")
    
    def render(self):
        print("Graphics: Rendering frame")

class AudioSystem:
    def initialize(self):
        print("AudioSystem: Initializing audio drivers")
    
    def play_sound(self, sound: str):
        print(f"AudioSystem: Playing {sound}")

class NetworkAdapter:
    def connect(self):
        print("NetworkAdapter: Establishing network connection")
    
    def download_updates(self):
        print("NetworkAdapter: Downloading system updates")

# Facade class
class ComputerFacade:
    def __init__(self):
        self.cpu = CPU()
        self.memory = Memory()
        self.hd = HardDrive()
        self.graphics = Graphics()
        self.audio = AudioSystem()
        self.network = NetworkAdapter()
    
    def start_computer(self):
        print("=== Starting Computer ===")
        self.cpu.freeze()
        
        boot_data = self.hd.read(0, 1024)
        self.memory.load(0, boot_data)
        
        self.cpu.jump(0)
        self.cpu.execute()
        
        self.graphics.initialize()
        self.audio.initialize()
        
        print("=== Computer Started Successfully ===")
    
    def shutdown_computer(self):
        print("=== Shutting Down Computer ===")
        self.audio.play_sound("shutdown_sound.wav")
        self.graphics.render()  # Final screen
        self.cpu.freeze()
        print("=== Computer Shut Down ===")
    
    def update_system(self):
        print("=== Updating System ===")
        self.network.connect()
        self.network.download_updates()
        print("=== System Updated ===")

# Usage
computer = ComputerFacade()

# Simple operations hiding complex subsystem interactions
computer.start_computer()
print()
computer.update_system()
print()
computer.shutdown_computer()

Architecture Diagram

Flyweight Pattern

The Flyweight Pattern minimizes memory usage by sharing common data among multiple objects . This pattern separates intrinsic state (shared) from extrinsic state (unique) to reduce memory consumption when dealing with large numbers of similar objects. refactoring.guru

When to Use

Applications spawning huge numbers of similar objects
Objects containing duplicate states that can be extracted and shared
Memory optimization for fine-grained objects

Python Implementation

 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
from typing import Dict, Optional
import weakref

# Flyweight interface
class CharacterFlyweight:
    def __init__(self, character: str, font: str, size: int):
        self.character = character  # Intrinsic state
        self.font = font           # Intrinsic state
        self.size = size           # Intrinsic state
    
    def render(self, x: int, y: int, color: str):
        """Render character with extrinsic state (position, color)"""
        print(f"Rendering '{self.character}' at ({x}, {y}) "
              f"in {self.font} font, size {self.size}, color {color}")

# Flyweight factory
class CharacterFlyweightFactory:
    _flyweights: Dict[tuple, CharacterFlyweight] = {}
    
    @classmethod
    def get_flyweight(cls, character: str, font: str, size: int) -> CharacterFlyweight:
        key = (character, font, size)
        
        if key not in cls._flyweights:
            cls._flyweights[key] = CharacterFlyweight(character, font, size)
            print(f"Created new flyweight for '{character}' in {font}, size {size}")
        
        return cls._flyweights[key]
    
    @classmethod
    def get_flyweight_count(cls) -> int:
        return len(cls._flyweights)

# Context class that uses flyweights
class Character:
    def __init__(self, char: str, x: int, y: int, color: str, font: str, size: int):
        self.flyweight = CharacterFlyweightFactory.get_flyweight(char, font, size)
        self.x = x        # Extrinsic state
        self.y = y        # Extrinsic state
        self.color = color # Extrinsic state
    
    def render(self):
        self.flyweight.render(self.x, self.y, self.color)

# Document class that manages characters
class Document:
    def __init__(self):
        self.characters = []
    
    def add_character(self, char: str, x: int, y: int, color: str, font: str, size: int):
        character = Character(char, x, y, color, font, size)
        self.characters.append(character)
    
    def render(self):
        print("=== Rendering Document ===")
        for char in self.characters:
            char.render()
        print(f"Total characters: {len(self.characters)}")
        print(f"Flyweight objects created: {CharacterFlyweightFactory.get_flyweight_count()}")

# Usage example
doc = Document()

# Adding text "HELLO WORLD" with repeated characters
text = "HELLO WORLD"
x_pos = 0

for char in text:
    if char != ' ':
        doc.add_character(char, x_pos, 100, "black", "Arial", 12)
    x_pos += 10

# Add more text with same characters
more_text = "HELLO PYTHON"
x_pos = 0

for char in more_text:
    if char != ' ':
        doc.add_character(char, x_pos, 150, "blue", "Arial", 12)
    x_pos += 10

# Render the document
doc.render()

Architecture Diagram

Proxy Pattern

The Proxy Pattern provides a surrogate or placeholder for another object to control access to it. This pattern can be used for security, lazy loading, remote access, or logging purposes. geeksforgeeks

When to Use

  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
121
122
123
from abc import ABC, abstractmethod
import time
from typing import Optional

# Subject interface
class DatabaseService(ABC):
    @abstractmethod
    def query(self, sql: str) -> str:
        pass
    
    @abstractmethod
    def execute(self, sql: str) -> bool:
        pass

# Real subject - expensive database connection
class RealDatabaseService(DatabaseService):
    def __init__(self):
        self._connect()
    
    def _connect(self):
        print("RealDatabaseService: Establishing database connection...")
        time.sleep(2)  # Simulate connection time
        print("RealDatabaseService: Connected to database")
    
    def query(self, sql: str) -> str:
        print(f"RealDatabaseService: Executing query: {sql}")
        time.sleep(1)  # Simulate query time
        return f"Result for: {sql}"
    
    def execute(self, sql: str) -> bool:
        print(f"RealDatabaseService: Executing command: {sql}")
        time.sleep(1)  # Simulate execution time
        return True

# Proxy with multiple functionalities
class DatabaseProxy(DatabaseService):
    def __init__(self, user_role: str):
        self._real_service: Optional[RealDatabaseService] = None
        self._user_role = user_role
        self._cache = {}
    
    def _get_real_service(self) -> RealDatabaseService:
        """Lazy initialization"""
        if self._real_service is None:
            print("DatabaseProxy: Creating real database service...")
            self._real_service = RealDatabaseService()
        return self._real_service
    
    def _check_access(self, operation: str) -> bool:
        """Security check"""
        if operation.upper().startswith(('DROP', 'DELETE', 'TRUNCATE')):
            return self._user_role == 'admin'
        return True
    
    def _log_operation(self, operation: str, sql: str):
        """Logging functionality"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        print(f"LOG [{timestamp}] User:{self._user_role} Operation:{operation} SQL:{sql}")
    
    def query(self, sql: str) -> str:
        # Security check
        if not self._check_access(sql):
            return "Access denied: Insufficient privileges"
        
        # Logging
        self._log_operation("QUERY", sql)
        
        # Caching
        if sql in self._cache:
            print("DatabaseProxy: Returning cached result")
            return self._cache[sql]
        
        # Lazy loading and delegation
        result = self._get_real_service().query(sql)
        
        # Cache the result
        self._cache[sql] = result
        
        return result
    
    def execute(self, sql: str) -> bool:
        # Security check
        if not self._check_access(sql):
            print("Access denied: Insufficient privileges for this operation")
            return False
        
        # Logging
        self._log_operation("EXECUTE", sql)
        
        # Clear cache for write operations
        if sql.upper().startswith(('INSERT', 'UPDATE', 'DELETE')):
            self._cache.clear()
            print("DatabaseProxy: Cache cleared due to write operation")
        
        # Delegate to real service
        return self._get_real_service().execute(sql)

# Usage example
print("=== Testing with Regular User ===")
user_proxy = DatabaseProxy("user")

# Query operations (should work)
result1 = user_proxy.query("SELECT * FROM users")
print(f"Result: {result1}\n")

# Cached query
result2 = user_proxy.query("SELECT * FROM users")
print(f"Cached Result: {result2}\n")

# Dangerous operation (should be denied)
user_proxy.execute("DROP TABLE users")
print()

print("=== Testing with Admin User ===")
admin_proxy = DatabaseProxy("admin")

# Admin can perform dangerous operations
admin_proxy.execute("DROP TABLE temp_data")
print()

# New query after write operation
result3 = admin_proxy.query("SELECT COUNT(*) FROM products")
print(f"Result: {result3}")

Architecture Diagram

Behavioral Design Patterns

Strategy Pattern

The Strategy pattern defines a family of algorithms, encapsulates each one, and makes them interchangeable. It lets algorithms vary independently from clients that use them. refactoring.guru

Use Cases

Payment processing systems with multiple payment methods
Sorting algorithms selection based on data size
Compression algorithms based on file types
Tax calculation strategies for different regions

Python Implementation

 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
from abc import ABC, abstractmethod
from typing import List

class PaymentStrategy(ABC):
    """Abstract strategy interface"""
    
    @abstractmethod
    def process_payment(self, amount: float) -> str:
        pass

class CreditCardStrategy(PaymentStrategy):
    """Concrete strategy for credit card payments"""
    
    def __init__(self, card_number: str, cvv: str):
        self.card_number = card_number
        self.cvv = cvv
    
    def process_payment(self, amount: float) -> str:
        return f"Paid ${amount:.2f} using Credit Card ending in {self.card_number[-4:]}"

class PayPalStrategy(PaymentStrategy):
    """Concrete strategy for PayPal payments"""
    
    def __init__(self, email: str):
        self.email = email
    
    def process_payment(self, amount: float) -> str:
        return f"Paid ${amount:.2f} using PayPal account {self.email}"

class BankTransferStrategy(PaymentStrategy):
    """Concrete strategy for bank transfers"""
    
    def __init__(self, account_number: str):
        self.account_number = account_number
    
    def process_payment(self, amount: float) -> str:
        return f"Paid ${amount:.2f} using Bank Transfer from account {self.account_number}"

class ShoppingCart:
    """Context class that uses the strategy"""
    
    def __init__(self):
        self.items: List[tuple] = []
        self.payment_strategy: PaymentStrategy = None
    
    def add_item(self, item: str, price: float):
        self.items.append((item, price))
    
    def set_payment_strategy(self, strategy: PaymentStrategy):
        self.payment_strategy = strategy
    
    def checkout(self) -> str:
        total = sum(price for _, price in self.items)
        if self.payment_strategy:
            return self.payment_strategy.process_payment(total)
        return "No payment strategy set"

# Usage example
cart = ShoppingCart()
cart.add_item("Laptop", 1200.00)
cart.add_item("Mouse", 25.00)

# Switch between different payment strategies
cart.set_payment_strategy(CreditCardStrategy("1234-5678-9012-3456", "123"))
print(cart.checkout())

cart.set_payment_strategy(PayPalStrategy("user@example.com"))
print(cart.checkout())

Diagram

Best Practices

Use when you have multiple ways to perform a task and want to switch between them dynamically
Favor composition over inheritance for algorithm selection
Keep strategies stateless when possible
Use factory pattern to create strategies based on configuration

Observer Pattern

The Observer pattern defines a one-to-many dependency between objects so that when one object changes state, all dependents are notified and updated automatically. viblo.asia

Use Cases

Model-View architectures (MVC, MVP, MVVM)
Event handling systems
Stock price monitoring
News subscription services
Real-time data feeds. blog.nashtechglobal.com

Python Implementation

  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
from typing import List, Any

class Observer(ABC):
    """Abstract observer interface"""
    
    @abstractmethod
    def update(self, subject: 'Subject', data: Any) -> None:
        pass

class Subject(ABC):
    """Abstract subject interface"""
    
    def __init__(self):
        self._observers: List[Observer] = []
    
    def subscribe(self, observer: Observer) -> None:
        self._observers.append(observer)
    
    def unsubscribe(self, observer: Observer) -> None:
        if observer in self._observers:
            self._observers.remove(observer)
    
    def notify(self, data: Any = None) -> None:
        for observer in self._observers:
            observer.update(self, data)

class StockPrice(Subject):
    """Concrete subject - stock price tracker"""
    
    def __init__(self, symbol: str, price: float):
        super().__init__()
        self._symbol = symbol
        self._price = price
    
    @property
    def price(self) -> float:
        return self._price
    
    @property
    def symbol(self) -> str:
        return self._symbol
    
    def set_price(self, new_price: float) -> None:
        self._price = new_price
        self.notify({"symbol": self._symbol, "price": new_price})

class StockDisplay(Observer):
    """Concrete observer - displays stock information"""
    
    def __init__(self, name: str):
        self.name = name
    
    def update(self, subject: Subject, data: Any) -> None:
        if isinstance(subject, StockPrice) and data:
            print(f"{self.name}: {data['symbol']} is now ${data['price']:.2f}")

class StockAlert(Observer):
    """Concrete observer - sends alerts for price changes"""
    
    def __init__(self, threshold: float):
        self.threshold = threshold
    
    def update(self, subject: Subject, data: Any) -> None:
        if isinstance(subject, StockPrice) and data:
            if data['price'] > self.threshold:
                print(f"ALERT: {data['symbol']} exceeded ${self.threshold:.2f}! Current: ${data['price']:.2f}")

class Portfolio(Observer):
    """Concrete observer - tracks portfolio value"""
    
    def __init__(self):
        self.holdings = {}
        self.total_value = 0.0
    
    def add_stock(self, symbol: str, shares: int):
        self.holdings[symbol] = shares
    
    def update(self, subject: Subject, data: Any) -> None:
        if isinstance(subject, StockPrice) and data:
            symbol = data['symbol']
            if symbol in self.holdings:
                shares = self.holdings[symbol]
                stock_value = shares * data['price']
                print(f"Portfolio: {shares} shares of {symbol} = ${stock_value:.2f}")

# Usage example
apple_stock = StockPrice("AAPL", 150.00)

# Create observers
display = StockDisplay("Main Dashboard")
alert = StockAlert(155.00)
portfolio = Portfolio()
portfolio.add_stock("AAPL", 100)

# Subscribe observers
apple_stock.subscribe(display)
apple_stock.subscribe(alert)
apple_stock.subscribe(portfolio)

# Trigger price changes
apple_stock.set_price(152.50)
apple_stock.set_price(157.00)

Diagram

Best Practices

Keep observers lightweight to avoid performance issues
Consider using weak references to prevent memory leaks
Implement proper error handling in observer updates
Use event-driven architectures for loose coupling
Consider async observers for non-blocking operations. refactoring.guru

Command Pattern

The Command pattern turns a request into a stand-alone object containing all information about the request. This transformation lets you pass requests as method arguments, delay execution, queue operations, and support undoable operations. refactoring.guru

Use Cases

GUI buttons and menu items
Macro recording and playback
Undo/Redo functionality
Transaction processing
Job scheduling and queuing
Remote procedure calls. geeksforgeeks

Python Implementation

  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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
from abc import ABC, abstractmethod
from typing import List, Optional
import time

class Command(ABC):
    """Abstract command interface"""
    
    @abstractmethod
    def execute(self) -> str:
        pass
    
    @abstractmethod
    def undo(self) -> str:
        pass

class Light:
    """Receiver class - the light that will be controlled"""
    
    def __init__(self, location: str):
        self.location = location
        self.is_on = False
        self.brightness = 0
    
    def turn_on(self) -> str:
        self.is_on = True
        self.brightness = 100
        return f"{self.location} light is ON (brightness: {self.brightness}%)"
    
    def turn_off(self) -> str:
        self.is_on = False
        self.brightness = 0
        return f"{self.location} light is OFF"
    
    def dim(self, level: int) -> str:
        if self.is_on:
            self.brightness = max(0, min(100, level))
            return f"{self.location} light dimmed to {self.brightness}%"
        return f"{self.location} light is OFF - cannot dim"

class LightOnCommand(Command):
    """Concrete command to turn light on"""
    
    def __init__(self, light: Light):
        self.light = light
        self.previous_state = False
        self.previous_brightness = 0
    
    def execute(self) -> str:
        self.previous_state = self.light.is_on
        self.previous_brightness = self.light.brightness
        return self.light.turn_on()
    
    def undo(self) -> str:
        if not self.previous_state:
            return self.light.turn_off()
        else:
            return self.light.dim(self.previous_brightness)

class LightOffCommand(Command):
    """Concrete command to turn light off"""
    
    def __init__(self, light: Light):
        self.light = light
        self.previous_state = False
        self.previous_brightness = 0
    
    def execute(self) -> str:
        self.previous_state = self.light.is_on
        self.previous_brightness = self.light.brightness
        return self.light.turn_off()
    
    def undo(self) -> str:
        if self.previous_state:
            self.light.is_on = True
            return self.light.dim(self.previous_brightness)
        return f"{self.light.location} light was already OFF"

class DimLightCommand(Command):
    """Concrete command to dim light"""
    
    def __init__(self, light: Light, level: int):
        self.light = light
        self.level = level
        self.previous_brightness = 0
    
    def execute(self) -> str:
        self.previous_brightness = self.light.brightness
        return self.light.dim(self.level)
    
    def undo(self) -> str:
        return self.light.dim(self.previous_brightness)

class MacroCommand(Command):
    """Command to execute multiple commands"""
    
    def __init__(self, commands: List[Command]):
        self.commands = commands
    
    def execute(self) -> str:
        results = []
        for command in self.commands:
            results.append(command.execute())
        return "Macro executed: " + "; ".join(results)
    
    def undo(self) -> str:
        results = []
        # Undo in reverse order
        for command in reversed(self.commands):
            results.append(command.undo())
        return "Macro undone: " + "; ".join(results)

class RemoteControl:
    """Invoker class - the remote control"""
    
    def __init__(self, slots: int = 7):
        self.on_commands: List[Optional[Command]] = [None] * slots
        self.off_commands: List[Optional[Command]] = [None] * slots
        self.last_command: Optional[Command] = None
        self.command_history: List[Command] = []
    
    def set_command(self, slot: int, on_command: Command, off_command: Command):
        self.on_commands[slot] = on_command
        self.off_commands[slot] = off_command
    
    def on_button_pressed(self, slot: int) -> str:
        if self.on_commands[slot]:
            result = self.on_commands[slot].execute()
            self.last_command = self.on_commands[slot]
            self.command_history.append(self.on_commands[slot])
            return result
        return f"No command set for slot {slot}"
    
    def off_button_pressed(self, slot: int) -> str:
        if self.off_commands[slot]:
            result = self.off_commands[slot].execute()
            self.last_command = self.off_commands[slot]
            self.command_history.append(self.off_commands[slot])
            return result
        return f"No command set for slot {slot}"
    
    def undo_button_pressed(self) -> str:
        if self.last_command:
            result = self.last_command.undo()
            self.last_command = None
            return f"Undo: {result}"
        return "No command to undo"
    
    def execute_macro(self, macro: MacroCommand) -> str:
        result = macro.execute()
        self.last_command = macro
        self.command_history.append(macro)
        return result

# Usage example
# Create devices (receivers)
living_room_light = Light("Living Room")
kitchen_light = Light("Kitchen")
bedroom_light = Light("Bedroom")

# Create commands
living_room_on = LightOnCommand(living_room_light)
living_room_off = LightOffCommand(living_room_light)
kitchen_on = LightOnCommand(kitchen_light)
kitchen_off = LightOffCommand(kitchen_light)
bedroom_dim = DimLightCommand(bedroom_light, 30)

# Create macro command
evening_routine = MacroCommand([
    LightOnCommand(living_room_light),
    LightOnCommand(kitchen_light),
    DimLightCommand(bedroom_light, 20)
])

# Create remote control (invoker)
remote = RemoteControl()

# Set up commands
remote.set_command(0, living_room_on, living_room_off)
remote.set_command(1, kitchen_on, kitchen_off)

# Use the remote
print(remote.on_button_pressed(0))    # Turn on living room light
print(remote.on_button_pressed(1))    # Turn on kitchen light
print(remote.undo_button_pressed())   # Undo last command

# Execute macro
print(remote.execute_macro(evening_routine))
print(remote.undo_button_pressed())   # Undo macro

Diagram

Best Practices

Keep commands simple and focused on single operations
Store necessary state for undo operations
Use macro commands for composite operations
Consider using command queues for batch processing
Implement proper error handling and rollback mechanisms

Chain of Responsibility Pattern

The Chain of Responsibility pattern lets you pass requests along a chain of handlers. Upon receiving a request, each handler decides either to process the request or pass it to the next handler in the chain.

Use Cases

Request processing systems (approvals, authentication)
Logging frameworks with different log levels
Exception handling hierarchies
GUI event handling
Middleware in web frameworks
Support ticket escalation systems

Python Implementation

  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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
from abc import ABC, abstractmethod
from enum import Enum
from typing import Optional, Dict, Any
import logging

class RequestType(Enum):
    BASIC = "BASIC"
    PREMIUM = "PREMIUM"
    VIP = "VIP"
    EMERGENCY = "EMERGENCY"

class SupportRequest:
    """Request object containing all necessary information"""
    
    def __init__(self, request_id: str, customer_type: RequestType, 
                 issue_severity: int, description: str, amount: float = 0.0):
        self.request_id = request_id
        self.customer_type = customer_type
        self.issue_severity = issue_severity  # 1-10 scale
        self.description = description
        self.amount = amount
        self.handled_by = None
        self.resolution = None

class SupportHandler(ABC):
    """Abstract handler interface"""
    
    def __init__(self):
        self._next_handler: Optional[SupportHandler] = None
    
    def set_next(self, handler: 'SupportHandler') -> 'SupportHandler':
        self._next_handler = handler
        return handler
    
    @abstractmethod
    def can_handle(self, request: SupportRequest) -> bool:
        pass
    
    @abstractmethod
    def handle_request(self, request: SupportRequest) -> str:
        pass
    
    def handle(self, request: SupportRequest) -> str:
        if self.can_handle(request):
            return self.handle_request(request)
        elif self._next_handler:
            return self._next_handler.handle(request)
        else:
            return f"Request {request.request_id} could not be handled by any support level"

class Level1Support(SupportHandler):
    """First level support - handles basic issues"""
    
    def can_handle(self, request: SupportRequest) -> bool:
        return (request.issue_severity <= 3 and 
                request.customer_type in [RequestType.BASIC, RequestType.PREMIUM])
    
    def handle_request(self, request: SupportRequest) -> str:
        request.handled_by = "Level 1 Support"
        request.resolution = f"Basic troubleshooting provided for {request.description}"
        return f"L1 Support resolved request {request.request_id}: {request.resolution}"

class Level2Support(SupportHandler):
    """Second level support - handles intermediate issues"""
    
    def can_handle(self, request: SupportRequest) -> bool:
        return (request.issue_severity <= 7 and 
                request.customer_type != RequestType.EMERGENCY)
    
    def handle_request(self, request: SupportRequest) -> str:
        request.handled_by = "Level 2 Support"
        request.resolution = f"Advanced technical solution provided for {request.description}"
        return f"L2 Support resolved request {request.request_id}: {request.resolution}"

class ManagerSupport(SupportHandler):
    """Manager level support - handles VIP and complex issues"""
    
    def can_handle(self, request: SupportRequest) -> bool:
        return (request.customer_type == RequestType.VIP or 
                request.issue_severity >= 8 or 
                request.amount > 10000)
    
    def handle_request(self, request: SupportRequest) -> str:
        request.handled_by = "Manager Support"
        if request.amount > 10000:
            request.resolution = f"Financial escalation handled: ${request.amount:.2f}"
        else:
            request.resolution = f"Executive-level resolution for {request.description}"
        return f"Manager resolved request {request.request_id}: {request.resolution}"

class DirectorSupport(SupportHandler):
    """Director level support - handles emergency and critical issues"""
    
    def can_handle(self, request: SupportRequest) -> bool:
        return (request.customer_type == RequestType.EMERGENCY or 
                request.issue_severity >= 9 or 
                request.amount > 50000)
    
    def handle_request(self, request: SupportRequest) -> str:
        request.handled_by = "Director Support"
        request.resolution = f"Critical issue resolved at director level: {request.description}"
        return f"Director resolved request {request.request_id}: {request.resolution}"

class SupportSystem:
    """Context class that sets up the chain"""
    
    def __init__(self):
        # Build the chain
        self.level1 = Level1Support()
        self.level2 = Level2Support()
        self.manager = ManagerSupport()
        self.director = DirectorSupport()
        
        # Set up the chain
        self.level1.set_next(self.level2).set_next(self.manager).set_next(self.director)
        
        self.processed_requests: List[SupportRequest] = []
    
    def process_request(self, request: SupportRequest) -> str:
        result = self.level1.handle(request)
        self.processed_requests.append(request)
        return result
    
    def get_stats(self) -> Dict[str, int]:
        stats = {}
        for request in self.processed_requests:
            handler = request.handled_by or "Unhandled"
            stats[handler] = stats.get(handler, 0) + 1
        return stats

# Usage example
support_system = SupportSystem()

# Create various types of requests
requests = [
    SupportRequest("REQ001", RequestType.BASIC, 2, "Password reset", 0),
    SupportRequest("REQ002", RequestType.PREMIUM, 5, "Integration issue", 0),
    SupportRequest("REQ003", RequestType.VIP, 4, "Performance optimization", 0),
    SupportRequest("REQ004", RequestType.BASIC, 9, "System outage", 0),
    SupportRequest("REQ005", RequestType.PREMIUM, 3, "Billing inquiry", 25000),
    SupportRequest("REQ006", RequestType.EMERGENCY, 10, "Security breach", 0),
]

# Process requests
print("=== Support Request Processing ===")
for request in requests:
    result = support_system.process_request(request)
    print(result)

print("\n=== Processing Statistics ===")
stats = support_system.get_stats()
for handler, count in stats.items():
    print(f"{handler}: {count} requests")

Diagram

Best Practices

Keep the chain as short as possible for performance
Ensure at least one handler can process any request
Consider using priority-based handling
Log chain traversal for debugging
Make handlers stateless and reusable

State Pattern

The State pattern allows an object to alter its behavior when its internal state changes. The object appears as if it changed its class. refactoring.guru

Use Cases

State machines (traffic lights, vending machines)
Game character states (idle, walking, jumping, attacking)
Document workflow (draft, review, published)
Connection states (connecting, connected, disconnected)
Order processing states

Python Implementation

  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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
from abc import ABC, abstractmethod
from enum import Enum
from typing import Optional
import time

class VendingMachineContext:
    """Context class representing a vending machine"""
    
    def __init__(self, inventory_count: int = 10):
        self.inventory_count = inventory_count
        self.inserted_money = 0.0
        self.item_price = 1.50
        self._state = NoMoneyState()
        self._state.set_context(self)
    
    def set_state(self, state: 'State'):
        print(f"State changed to: {state.__class__.__name__}")
        self._state = state
        self._state.set_context(self)
    
    def insert_money(self, amount: float) -> str:
        return self._state.insert_money(amount)
    
    def select_item(self) -> str:
        return self._state.select_item()
    
    def dispense_item(self) -> str:
        return self._state.dispense_item()
    
    def cancel_transaction(self) -> str:
        return self._state.cancel_transaction()
    
    def get_status(self) -> str:
        return (f"State: {self._state.__class__.__name__}, "
                f"Money: ${self.inserted_money:.2f}, "
                f"Inventory: {self.inventory_count}")

class State(ABC):
    """Abstract state interface"""
    
    def __init__(self):
        self._context: Optional[VendingMachineContext] = None
    
    def set_context(self, context: VendingMachineContext):
        self._context = context
    
    @abstractmethod
    def insert_money(self, amount: float) -> str:
        pass
    
    @abstractmethod
    def select_item(self) -> str:
        pass
    
    @abstractmethod
    def dispense_item(self) -> str:
        pass
    
    @abstractmethod
    def cancel_transaction(self) -> str:
        pass

class NoMoneyState(State):
    """State when no money has been inserted"""
    
    def insert_money(self, amount: float) -> str:
        if amount <= 0:
            return "Please insert a valid amount"
        
        self._context.inserted_money += amount
        
        if self._context.inserted_money >= self._context.item_price:
            self._context.set_state(HasMoneyState())
            return f"${amount:.2f} inserted. You can now select an item."
        else:
            needed = self._context.item_price - self._context.inserted_money
            return f"${amount:.2f} inserted. Please insert ${needed:.2f} more."
    
    def select_item(self) -> str:
        return "Please insert money first"
    
    def dispense_item(self) -> str:
        return "Please insert money and select an item first"
    
    def cancel_transaction(self) -> str:
        if self._context.inserted_money > 0:
            refund = self._context.inserted_money
            self._context.inserted_money = 0
            return f"Transaction cancelled. ${refund:.2f} refunded."
        return "No transaction to cancel"

class HasMoneyState(State):
    """State when sufficient money has been inserted"""
    
    def insert_money(self, amount: float) -> str:
        if amount > 0:
            self._context.inserted_money += amount
            return f"${amount:.2f} added. Total: ${self._context.inserted_money:.2f}"
        return "Invalid amount"
    
    def select_item(self) -> str:
        if self._context.inventory_count <= 0:
            self._context.set_state(SoldOutState())
            return "Sorry, item is sold out"
        
        self._context.set_state(DispensingState())
        return "Item selected. Dispensing..."
    
    def dispense_item(self) -> str:
        return "Please select an item first"
    
    def cancel_transaction(self) -> str:
        refund = self._context.inserted_money
        self._context.inserted_money = 0
        self._context.set_state(NoMoneyState())
        return f"Transaction cancelled. ${refund:.2f} refunded."

class DispensingState(State):
    """State when item is being dispensed"""
    
    def insert_money(self, amount: float) -> str:
        return "Please wait, dispensing item"
    
    def select_item(self) -> str:
        return "Already dispensing an item"
    
    def dispense_item(self) -> str:
        self._context.inventory_count -= 1
        
        # Calculate change
        change = self._context.inserted_money - self._context.item_price
        self._context.inserted_money = 0
        
        # Determine next state
        if self._context.inventory_count <= 0:
            self._context.set_state(SoldOutState())
        else:
            self._context.set_state(NoMoneyState())
        
        change_msg = f" Change: ${change:.2f}" if change > 0 else ""
        return f"Item dispensed!{change_msg}"
    
    def cancel_transaction(self) -> str:
        return "Cannot cancel during dispensing"

class SoldOutState(State):
    """State when machine is sold out"""
    
    def insert_money(self, amount: float) -> str:
        return "Machine is sold out. Money not accepted."
    
    def select_item(self) -> str:
        return "Machine is sold out"
    
    def dispense_item(self) -> str:
        return "Machine is sold out"
    
    def cancel_transaction(self) -> str:
        if self._context.inserted_money > 0:
            refund = self._context.inserted_money
            self._context.inserted_money = 0
            return f"Machine sold out. ${refund:.2f} refunded."
        return "Machine is sold out"

# Usage example
def simulate_vending_machine():
    print("=== Vending Machine Simulation ===")
    machine = VendingMachineContext(inventory_count=3)
    
    print(f"Initial status: {machine.get_status()}")
    print()
    
    # Test scenarios
    test_scenarios = [
        ("Try to select without money", lambda: machine.select_item()),
        ("Insert insufficient money", lambda: machine.insert_money(1.00)),
        ("Insert more money", lambda: machine.insert_money(0.75)),
        ("Select item", lambda: machine.select_item()),
        ("Dispense item", lambda: machine.dispense_item()),
        ("Insert money for second item", lambda: machine.insert_money(2.00)),
        ("Select and dispense", lambda: (machine.select_item(), machine.dispense_item())),
        ("Buy last item", lambda: (machine.insert_money(1.50), machine.select_item(), machine.dispense_item())),
        ("Try to buy when sold out", lambda: machine.insert_money(1.50)),
    ]
    
    for description, action in test_scenarios:
        print(f"--- {description} ---")
        try:
            result = action()
            if isinstance(result, tuple):
                for r in result:
                    print(r)
            else:
                print(result)
        except Exception as e:
            print(f"Error: {e}")
        
        print(f"Status: {machine.get_status()}")
        print()

simulate_vending_machine()

Diagram

Best Practices

Keep states focused on specific behaviors
Minimize state transitions and coupling
Use enums or constants for state identification
Handle invalid transitions gracefully
Consider using state machines for complex workflows

Template Method Pattern

The Template Method pattern defines the skeleton of an algorithm in a superclass but lets subclasses override specific steps without changing the algorithm’s structure. refactoring.guru

Use Cases

Data processing pipelines with common steps
Test frameworks with setup/teardown methods
Document generation with different formats
Authentication workflows
Cooking recipes with variations
Build processes

Python Implementation

  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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import json
import csv
import xml.etree.ElementTree as ET
from io import StringIO

class DataProcessor(ABC):
    """Abstract template class defining the data processing algorithm"""
    
    def process_data(self, file_path: str) -> str:
        """Template method defining the algorithm structure"""
        print(f"Starting data processing for: {file_path}")
        
        # Step 1: Read data
        raw_data = self.read_data(file_path)
        
        # Step 2: Parse data
        parsed_data = self.parse_data(raw_data)
        
        # Step 3: Validate data (hook method - optional)
        if self.should_validate():
            validated_data = self.validate_data(parsed_data)
        else:
            validated_data = parsed_data
        
        # Step 4: Transform data
        transformed_data = self.transform_data(validated_data)
        
        # Step 5: Generate output
        output = self.generate_output(transformed_data)
        
        # Step 6: Post-processing hook (optional)
        self.post_process()
        
        print("Data processing completed")
        return output
    
    @abstractmethod
    def read_data(self, file_path: str) -> str:
        """Abstract method - must be implemented by subclasses"""
        pass
    
    @abstractmethod
    def parse_data(self, raw_data: str) -> List[Dict[str, Any]]:
        """Abstract method - must be implemented by subclasses"""
        pass
    
    def validate_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Concrete method with default implementation"""
        print("Validating data...")
        validated = []
        for record in data:
            if self.is_valid_record(record):
                validated.append(record)
            else:
                print(f"Invalid record skipped: {record}")
        return validated
    
    def is_valid_record(self, record: Dict[str, Any]) -> bool:
        """Hook method for record validation"""
        return len(record) > 0 and all(value is not None for value in record.values())
    
    @abstractmethod
    def transform_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Abstract method for data transformation"""
        pass
    
    @abstractmethod
    def generate_output(self, data: List[Dict[str, Any]]) -> str:
        """Abstract method for output generation"""
        pass
    
    def should_validate(self) -> bool:
        """Hook method - can be overridden to skip validation"""
        return True
    
    def post_process(self) -> None:
        """Hook method for post-processing"""
        pass

class JSONDataProcessor(DataProcessor):
    """Concrete processor for JSON data"""
    
    def read_data(self, file_path: str) -> str:
        print("Reading JSON file...")
        # Simulate reading a JSON file
        sample_data = '''[
            {"id": 1, "name": "Alice", "age": 30, "salary": 70000},
            {"id": 2, "name": "Bob", "age": 25, "salary": 60000},
            {"id": 3, "name": "Charlie", "age": 35, "salary": 80000}
        ]'''
        return sample_data
    
    def parse_data(self, raw_data: str) -> List[Dict[str, Any]]:
        print("Parsing JSON data...")
        return json.loads(raw_data)
    
    def transform_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        print("Transforming JSON data...")
        transformed = []
        for record in data:
            # Add calculated fields
            record['annual_salary'] = record.get('salary', 0) * 12
            record['age_category'] = 'Young' if record.get('age', 0) < 30 else 'Experienced'
            transformed.append(record)
        return transformed
    
    def generate_output(self, data: List[Dict[str, Any]]) -> str:
        print("Generating JSON output...")
        return json.dumps(data, indent=2)
    
    def post_process(self) -> None:
        print("JSON processing completed - sending notification...")

class CSVDataProcessor(DataProcessor):
    """Concrete processor for CSV data"""
    
    def read_data(self, file_path: str) -> str:
        print("Reading CSV file...")
        # Simulate reading a CSV file
        sample_data = '''id,name,age,salary
1,Alice,30,70000
2,Bob,25,60000
3,Charlie,35,80000
4,Diana,,75000'''
        return sample_data
    
    def parse_data(self, raw_data: str) -> List[Dict[str, Any]]:
        print("Parsing CSV data...")
        lines = raw_data.strip().split('\n')
        headers = lines[0].split(',')
        
        data = []
        for line in lines[1:]:
            values = line.split(',')
            record = {}
            for i, header in enumerate(headers):
                value = values[i] if i < len(values) and values[i] else None
                # Convert numeric fields
                if header in ['id', 'age', 'salary'] and value:
                    try:
                        record[header] = int(value)
                    except ValueError:
                        record[header] = None
                else:
                    record[header] = value
            data.append(record)
        return data
    
    def is_valid_record(self, record: Dict[str, Any]) -> bool:
        """Override validation for CSV-specific rules"""
        return (record.get('id') is not None and 
                record.get('name') is not None and 
                record.get('salary') is not None)
    
    def transform_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        print("Transforming CSV data...")
        transformed = []
        for record in data:
            # Calculate tax bracket
            salary = record.get('salary', 0)
            if salary < 50000:
                record['tax_bracket'] = 'Low'
            elif salary < 80000:
                record['tax_bracket'] = 'Medium'
            else:
                record['tax_bracket'] = 'High'
            transformed.append(record)
        return transformed
    
    def generate_output(self, data: List[Dict[str, Any]]) -> str:
        print("Generating CSV output...")
        if not data:
            return ""
        
        headers = list(data[0].keys())
        output = StringIO()
        writer = csv.DictWriter(output, fieldnames=headers)
        writer.writeheader()
        writer.writerows(data)
        return output.getvalue()

class XMLDataProcessor(DataProcessor):
    """Concrete processor for XML data"""
    
    def __init__(self):
        self.validate_xml = True
    
    def read_data(self, file_path: str) -> str:
        print("Reading XML file...")
        # Simulate reading an XML file
        sample_data = '''<?xml version="1.0"?>
        <employees>
            <employee>
                <id>1</id>
                <name>Alice</name>
                <age>30</age>
                <salary>70000</salary>
            </employee>
            <employee>
                <id>2</id>
                <name>Bob</name>
                <age>25</age>
                <salary>60000</salary>
            </employee>
        </employees>'''
        return sample_data
    
    def parse_data(self, raw_data: str) -> List[Dict[str, Any]]:
        print("Parsing XML data...")
        root = ET.fromstring(raw_data)
        data = []
        
        for employee in root.findall('employee'):
            record = {}
            for child in employee:
                if child.tag in ['id', 'age', 'salary']:
                    try:
                        record[child.tag] = int(child.text) if child.text else None
                    except ValueError:
                        record[child.tag] = None
                else:
                    record[child.tag] = child.text
            data.append(record)
        return data
    
    def transform_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        print("Transforming XML data...")
        for record in data:
            # Add employee status
            age = record.get('age', 0)
            if age < 25:
                record['status'] = 'Junior'
            elif age < 40:
                record['status'] = 'Senior'
            else:
                record['status'] = 'Expert'
        return data
    
    def generate_output(self, data: List[Dict[str, Any]]) -> str:
        print("Generating XML output...")
        root = ET.Element("employees")
        
        for record in data:
            employee = ET.SubElement(root, "employee")
            for key, value in record.items():
                elem = ET.SubElement(employee, key)
                elem.text = str(value) if value is not None else ""
        
        return ET.tostring(root, encoding='unicode')
    
    def should_validate(self) -> bool:
        return self.validate_xml

# Usage example
def demonstrate_template_method():
    print("=== Template Method Pattern Demonstration ===\n")
    
    processors = [
        ("JSON Processor", JSONDataProcessor(), "data.json"),
        ("CSV Processor", CSVDataProcessor(), "data.csv"),
        ("XML Processor", XMLDataProcessor(), "data.xml"),
    ]
    
    for name, processor, filename in processors:
        print(f"{'='*50}")
        print(f"Processing with {name}")
        print(f"{'='*50}")
        
        try:
            result = processor.process_data(filename)
            print(f"\nOutput preview:")
            print(result[:200] + "..." if len(result) > 200 else result)
        except Exception as e:
            print(f"Error processing {filename}: {e}")
        
        print("\n")

demonstrate_template_method()

Diagram

Best Practices

Keep the template method final/non-overridable
Provide meaningful default implementations for hook methods
Use clear naming conventions for abstract vs hook methods
Document the algorithm steps and extension points
Minimize the number of abstract methods

Mediator Pattern

The Mediator pattern defines how a set of objects interact with each other. Instead of objects communicating directly, they communicate through a central mediator object. refactoring.guru

Use Cases

GUI dialog boxes with interdependent controls
Chat room systems
Air traffic control systems
Workflow management systems
Event broadcasting systems
Component communication in complex UIs

Python Implementation

  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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum

class MessageType(Enum):
    TEXT = "TEXT"
    SYSTEM = "SYSTEM"
    PRIVATE = "PRIVATE"
    BROADCAST = "BROADCAST"

class ChatMessage:
    """Represents a chat message"""
    
    def __init__(self, sender: str, content: str, msg_type: MessageType = MessageType.TEXT):
        self.sender = sender
        self.content = content
        self.msg_type = msg_type
        self.timestamp = datetime.now()
        self.recipients: List[str] = []

class Mediator(ABC):
    """Abstract mediator interface"""
    
    @abstractmethod
    def send_message(self, message: ChatMessage, sender: 'User') -> None:
        pass
    
    @abstractmethod
    def add_user(self, user: 'User') -> None:
        pass
    
    @abstractmethod
    def remove_user(self, user: 'User') -> None:
        pass

class User(ABC):
    """Abstract user/colleague class"""
    
    def __init__(self, name: str, mediator: Optional[Mediator] = None):
        self.name = name
        self.mediator = mediator
        self.message_history: List[ChatMessage] = []
        self.is_online = True
    
    def set_mediator(self, mediator: Mediator):
        self.mediator = mediator
    
    def send_message(self, content: str, msg_type: MessageType = MessageType.TEXT) -> None:
        if self.mediator and self.is_online:
            message = ChatMessage(self.name, content, msg_type)
            self.mediator.send_message(message, self)
    
    def send_private_message(self, recipient: str, content: str) -> None:
        if self.mediator and self.is_online:
            message = ChatMessage(self.name, content, MessageType.PRIVATE)
            message.recipients = [recipient]
            self.mediator.send_message(message, self)
    
    @abstractmethod
    def receive_message(self, message: ChatMessage) -> None:
        pass
    
    def go_offline(self):
        self.is_online = False
        if self.mediator:
            system_msg = ChatMessage("System", f"{self.name} went offline", MessageType.SYSTEM)
            self.mediator.send_message(system_msg, self)
    
    def go_online(self):
        self.is_online = True
        if self.mediator:
            system_msg = ChatMessage("System", f"{self.name} came online", MessageType.SYSTEM)
            self.mediator.send_message(system_msg, self)

class RegularUser(User):
    """Regular user implementation"""
    
    def receive_message(self, message: ChatMessage) -> None:
        if not self.is_online and message.msg_type != MessageType.SYSTEM:
            return  # Offline users don't receive messages (except system messages)
        
        self.message_history.append(message)
        timestamp = message.timestamp.strftime("%H:%M:%S")
        
        if message.msg_type == MessageType.PRIVATE:
            print(f"[{timestamp}] (Private) {message.sender} -> {self.name}: {message.content}")
        elif message.msg_type == MessageType.SYSTEM:
            print(f"[{timestamp}] (System): {message.content}")
        else:
            print(f"[{timestamp}] {message.sender}: {message.content}")

class ModeratorUser(User):
    """Moderator user with special privileges"""
    
    def __init__(self, name: str, mediator: Optional[Mediator] = None):
        super().__init__(name, mediator)
        self.warnings_issued: Dict[str, int] = {}
    
    def receive_message(self, message: ChatMessage) -> None:
        self.message_history.append(message)
        timestamp = message.timestamp.strftime("%H:%M:%S")
        
        if message.msg_type == MessageType.PRIVATE:
            print(f"[{timestamp}] (Private) {message.sender} -> {self.name}: {message.content}")
        elif message.msg_type == MessageType.SYSTEM:
            print(f"[{timestamp}] (System): {message.content}")
        else:
            print(f"[{timestamp}] [MOD] {message.sender}: {message.content}")
    
    def issue_warning(self, username: str, reason: str) -> None:
        self.warnings_issued[username] = self.warnings_issued.get(username, 0) + 1
        warning_count = self.warnings_issued[username]
        
        warning_msg = f"Warning #{warning_count} issued to {username}: {reason}"
        if self.mediator:
            message = ChatMessage(f"Moderator {self.name}", warning_msg, MessageType.SYSTEM)
            self.mediator.send_message(message, self)
    
    def broadcast_announcement(self, content: str) -> None:
        if self.mediator:
            message = ChatMessage(f"Moderator {self.name}", f"ANNOUNCEMENT: {content}", MessageType.BROADCAST)
            self.mediator.send_message(message, self)

class ChatRoom(Mediator):
    """Concrete mediator - chat room implementation"""
    
    def __init__(self, name: str):
        self.name = name
        self.users: List[User] = []
        self.banned_words = ["spam", "toxic", "inappropriate"]
        self.message_log: List[ChatMessage] = []
    
    def add_user(self, user: User) -> None:
        if user not in self.users:
            self.users.append(user)
            user.set_mediator(self)
            
            # Send welcome message
            welcome_msg = ChatMessage("System", f"{user.name} joined the chat room", MessageType.SYSTEM)
            self.send_message(welcome_msg, user)
    
    def remove_user(self, user: User) -> None:
        if user in self.users:
            self.users.remove(user)
            
            # Send goodbye message
            goodbye_msg = ChatMessage("System", f"{user.name} left the chat room", MessageType.SYSTEM)
            self.send_message(goodbye_msg, user)
    
    def send_message(self, message: ChatMessage, sender: User) -> None:
        # Log all messages
        self.message_log.append(message)
        
        # Content filtering
        if self._contains_banned_words(message.content):
            warning = ChatMessage("System", f"Message from {sender.name} was blocked due to inappropriate content", MessageType.SYSTEM)
            sender.receive_message(warning)
            return
        
        # Route message based on type
        if message.msg_type == MessageType.PRIVATE:
            self._send_private_message(message)
        elif message.msg_type == MessageType.BROADCAST:
            self._send_broadcast_message(message)
        elif message.msg_type == MessageType.SYSTEM:
            self._send_system_message(message)
        else:
            self._send_public_message(message, sender)
    
    def _contains_banned_words(self, content: str) -> bool:
        return any(word.lower() in content.lower() for word in self.banned_words)
    
    def _send_private_message(self, message: ChatMessage) -> None:
        for recipient_name in message.recipients:
            recipient = self._find_user(recipient_name)
            if recipient and recipient.is_online:
                recipient.receive_message(message)
    
    def _send_broadcast_message(self, message: ChatMessage) -> None:
        for user in self.users:
            user.receive_message(message)
    
    def _send_system_message(self, message: ChatMessage) -> None:
        for user in self.users:
            user.receive_message(message)
    
    def _send_public_message(self, message: ChatMessage, sender: User) -> None:
        for user in self.users:
            if user != sender:  # Don't echo message back to sender
                user.receive_message(message)
    
    def _find_user(self, username: str) -> Optional[User]:
        for user in self.users:
            if user.name == username:
                return user
        return None
    
    def get_online_users(self) -> List[str]:
        return [user.name for user in self.users if user.is_online]
    
    def get_chat_statistics(self) -> Dict[str, Any]:
        total_messages = len(self.message_log)
        message_types = {}
        for msg in self.message_log:
            msg_type = msg.msg_type.value
            message_types[msg_type] = message_types.get(msg_type, 0) + 1
        
        return {
            "total_users": len(self.users),
            "online_users": len([u for u in self.users if u.is_online]),
            "total_messages": total_messages,
            "message_types": message_types
        }

# Usage example
def demonstrate_chat_room():
    print("=== Chat Room Mediator Pattern Demonstration ===\n")
    
    # Create chat room (mediator)
    chat_room = ChatRoom("General Chat")
    
    # Create users
    alice = RegularUser("Alice")
    bob = RegularUser("Bob")
    charlie = RegularUser("Charlie")
    moderator = ModeratorUser("ModeratorMike")
    
    # Add users to chat room
    chat_room.add_user(alice)
    chat_room.add_user(bob)
    chat_room.add_user(charlie)
    chat_room.add_user(moderator)
    
    print("\n--- Public Messages ---")
    alice.send_message("Hello everyone!")
    bob.send_message("Hi Alice! How's everyone doing?")
    charlie.send_message("Great to be here!")
    
    print("\n--- Private Messages ---")
    alice.send_private_message("Bob", "Can we chat privately?")
    bob.send_private_message("Alice", "Sure, what's up?")
    
    print("\n--- Moderator Actions ---")
    moderator.broadcast_announcement("Welcome to the chat room! Please be respectful.")
    moderator.issue_warning("Charlie", "Please keep messages appropriate")
    
    print("\n--- User Going Offline ---")
    charlie.go_offline()
    alice.send_message("Did Charlie leave?")
    
    print("\n--- Content Filtering ---")
    bob.send_message("This is spam content")  # Will be blocked
    
    print("\n--- Chat Statistics ---")
    stats = chat_room.get_chat_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")
    
    print(f"\nOnline users: {', '.join(chat_room.get_online_users())}")

demonstrate_chat_room()

Diagram

Concurrency/Reactive Design Patterns

Use Case

Optimal for CPU-intensive tasks or I/O-bound tasks with blocking operations, where a fixed (or bounded) number of threads handle incoming work. Examples: batch processing, parallel computation, bulk file I/O.

Best Practices

Size your pool appropriately (CPU-bound: ≈ CPU cores; I/O-bound: higher).
Use futures for result retrieval and error handling.
Avoid mutable shared state; prefer worker isolation.
Graceful shutdown: Always use a context manager or explicit shutdown.

Python Example

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import concurrent.futures
import time

def task(n):
    print(f"Task {n} started")
    time.sleep(1)
    print(f"Task {n} finished")
    return n * n

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    results = [f.result() for f in futures]
    print("Results:", results)

Diagram

Actor Model

Ideal for distributed systems, message-driven services, and stateful processing where actors communicate solely via asynchronous message passing. Examples: chat services, real-time analytics, distributed workflows.

Best Practices

Isolate state: Each actor manages its own state; no shared mutable state.
Use message queues: Actors communicate through queues/channels, never directly.
Handle lifecycle: Manage startup, shutdown, and error recovery.
Leverage asyncio: For high-throughput, non-blocking message processing.

Python Example

 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
import asyncio

class Actor:
    def __init__(self):
        self.queue = asyncio.Queue()

    async def send(self, message):
        await self.queue.put(message)

    async def receive(self):
        while True:
            message = await self.queue.get()
            if message == "STOP":
                break
            await self.handle(message)

    async def handle(self, message):
        print(f"Handling message: {message}")

async def main():
    actor = Actor()
    asyncio.create_task(actor.receive())
    await actor.send("Hello")
    await actor.send("World")
    await actor.send("STOP")

asyncio.run(main())

Diagram

Reactor Pattern

Use Case

Best for event-driven, I/O-heavy applications (e.g., servers, proxies, GUI frameworks). The reactor demultiplexes events and dispatches them to handlers—often a single-threaded event loop.

Best Practices

Register handlers: For each I/O resource (socket, file, timer).
Non-blocking handlers: Keep handlers short and non-blocking; offload heavy work to threads.
Single-threaded core: The reactor loop should rarely block.
Error isolation: Handle errors in handlers to avoid crashing the reactor.

Python Example

 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
import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock):
    conn, addr = sock.accept()
    print(f"Connection from {addr}")
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn):
    data = conn.recv(1024)
    if data:
        print(f"Received: {data.decode()}")
        conn.send(data)  # Echo back
    else:
        sel.unregister(conn)
        conn.close()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('localhost', 12345))
sock.listen()
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)

print("Server running on localhost:12345")
while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj)