返回文章列表

Python設計模式開發可擴充套件應用

深入探討 Python 設計模式的理論與實踐,涵蓋建立型、結構型和行為型三大類別模式。透過完整程式碼範例展示 Singleton、Factory、Strategy、Observer 等經典模式在 Python 中的實作方式,並探討多執行緒環境下的執行緒安全設計、元類別應用與模式組合策略,協助開發者建構可維護、可擴充的企業級應用程式。

軟體工程 程式設計

軟體工程領域中,設計模式扮演著承先啟後的關鍵角色。這些經過時間驗證的解決方案,將抽象的設計原則具體化為可重複使用的程式碼結構,使開發者能夠站在前人的肩膀上,有效率地解決複雜的軟體設計問題。Python 語言以其優雅的語法和動態特性聞名,為設計模式的實踐提供了絕佳的實驗場域。本文將深入探討設計模式在 Python 中的應用,從理論基礎到實務實作,協助讀者建構可維護、可擴充的應用程式。

設計模式的本質與價值

設計模式的概念源自建築學領域,由 Christopher Alexander 在 1970 年代提出。他認為優秀的建築設計應該遵循某些可重複的模式,這些模式能夠解決特定情境下的設計問題。這個概念後來被軟體工程師採納,並在 1994 年由 Erich Gamma、Richard Helm、Ralph Johnson 和 John Vlissides 四位作者(後被稱為 Gang of Four,簡稱 GoF)正式整理成《Design Patterns: Elements of Reusable Object-Oriented Software》一書,奠定了現代設計模式的基礎。

設計模式之所以重要,在於它們提供了一套經過驗證的解決方案框架。當開發者面對複雜的軟體設計問題時,設計模式能夠提供明確的指引,避免重複發明輪子。更重要的是,設計模式建立了開發團隊之間的共同語言,當一位工程師說「這裡我們使用 Observer 模式」時,團隊成員立刻就能理解其設計意圖和實作方式。

從軟體工程的角度來看,設計模式體現了幾個核心原則。首先是關注點分離原則,每個模式都專注於解決特定類型的問題,使系統各部分職責明確。其次是開閉原則,許多模式的設計目標是讓系統對擴充開放、對修改封閉,這意味著可以透過新增程式碼來擴充功能,而不需要修改現有程式碼。第三是依賴反轉原則,高層模組不應該依賴低層模組,兩者都應該依賴抽象。這些原則共同構成了設計模式的理論基礎。

Python 作為一種動態型別語言,在實踐設計模式時有其獨特優勢。Python 的 duck typing 機制意味著物件的類型由其行為決定,而非繼承關係,這使得許多模式的實作更加簡潔。Python 的一級函式特性讓函式可以作為參數傳遞或儲存在資料結構中,這為某些模式提供了更優雅的實作方式。此外,Python 的元類別系統和裝飾器機制也為模式實作開闢了新的可能性。

建立型模式深入解析

建立型模式關注物件的建立機制,旨在將物件的建立過程與使用過程分離。這類模式的核心目標是增加系統的靈活性,讓程式碼能夠針對抽象介面編程,而非具體類別。透過適當運用建立型模式,系統可以在執行時期決定要建立哪種物件,或者將物件的建立委託給其他元件處理。

Singleton 模式的多種實作策略

Singleton 模式確保一個類別只有一個實例,並提供全域存取點。這個模式看似簡單,但在 Python 中有多種實作方式,每種方式都有其適用情境和權衡考量。

最基本的 Singleton 實作利用 Python 的 __new__ 方法來控制實例的建立過程:

class BasicSingleton:
    """
    基礎 Singleton 模式實作

    這個實作利用類別變數 _instance 來儲存唯一的實例
    __new__ 方法在物件建立時被呼叫,優先於 __init__
    透過檢查 _instance 是否存在來決定是否建立新實例
    """

    # 類別變數,用於儲存唯一的實例
    # 初始值為 None,表示尚未建立實例
    _instance = None

    def __new__(cls):
        """
        控制實例建立的特殊方法

        Args:
            cls: 當前類別的參考

        Returns:
            唯一的類別實例
        """
        # 檢查是否已經存在實例
        if cls._instance is None:
            # 如果不存在,呼叫父類別的 __new__ 建立新實例
            cls._instance = super(BasicSingleton, cls).__new__(cls)
            # 可以在這裡進行一次性的初始化
            cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        """
        初始化方法

        注意:每次取得實例時 __init__ 都會被呼叫
        因此需要使用旗標來確保真正的初始化只執行一次
        """
        # 使用旗標避免重複初始化
        if not self._initialized:
            # 執行實際的初始化邏輯
            self._data = {}
            self._initialized = True
            print("Singleton 實例初始化完成")

    def set_data(self, key, value):
        """
        設定資料的方法

        Args:
            key: 資料鍵
            value: 資料值
        """
        self._data[key] = value

    def get_data(self, key):
        """
        取得資料的方法

        Args:
            key: 資料鍵

        Returns:
            對應的資料值,若不存在則傳回 None
        """
        return self._data.get(key)

# 使用範例
if __name__ == "__main__":
    # 第一次建立實例
    singleton1 = BasicSingleton()
    singleton1.set_data("config", "production")

    # 第二次取得實例(實際上是同一個實例)
    singleton2 = BasicSingleton()

    # 驗證兩個變數指向同一個實例
    print(f"singleton1 is singleton2: {singleton1 is singleton2}")
    # 輸出:True

    # 驗證資料共享
    print(f"透過 singleton2 取得資料: {singleton2.get_data('config')}")
    # 輸出:production

在多執行緒環境中,上述基本實作可能會遇到競爭條件問題。當多個執行緒同時檢查 _instance is None 時,可能會建立多個實例。為了解決這個問題,需要使用鎖定機制:

import threading
from typing import Any

class ThreadSafeSingleton:
    """
    執行緒安全的 Singleton 模式實作

    使用元類別和鎖定機制確保在多執行緒環境中
    只會建立一個實例
    """

    # 類別變數
    _instance = None
    _lock = threading.Lock()  # 建立鎖定物件

    def __new__(cls, *args, **kwargs):
        """
        執行緒安全的實例建立方法

        使用雙重檢查鎖定模式(Double-Checked Locking)
        這個模式先檢查實例是否存在,如果不存在才取得鎖定
        取得鎖定後再次檢查,確保只有一個執行緒建立實例
        """
        # 第一次檢查(不需要鎖定)
        # 如果實例已存在,直接傳回,避免不必要的鎖定開銷
        if cls._instance is None:
            # 取得鎖定
            with cls._lock:
                # 第二次檢查(已取得鎖定)
                # 防止多個執行緒同時通過第一次檢查
                if cls._instance is None:
                    # 建立實例
                    cls._instance = super().__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        """
        執行緒安全的初始化方法
        """
        # 同樣需要鎖定來保護初始化過程
        with self._lock:
            if not self._initialized:
                self._data = {}
                self._initialized = True

class SingletonMeta(type):
    """
    使用元類別實作 Singleton 模式

    元類別是類別的類別,透過元類別可以控制類別的建立過程
    這是一種更 Pythonic 的 Singleton 實作方式
    """

    # 儲存所有 Singleton 類別的實例
    _instances = {}
    # 每個類別一個鎖定,避免不同類別之間互相阻塞
    _lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        """
        當類別被呼叫(即嘗試建立實例)時執行

        這個方法攔截了一般的實例建立流程
        確保每個類別只有一個實例
        """
        # 使用鎖定確保執行緒安全
        with cls._lock:
            # 檢查這個類別是否已經有實例
            if cls not in cls._instances:
                # 呼叫父類別(type)的 __call__ 來建立實例
                instance = super().__call__(*args, **kwargs)
                cls._instances[cls] = instance
        return cls._instances[cls]

class DatabaseConnection(metaclass=SingletonMeta):
    """
    使用元類別 Singleton 的資料庫連線類別

    這個類別確保整個應用程式只有一個資料庫連線實例
    避免資源浪費和連線管理問題
    """

    def __init__(self, host="localhost", port=5432, database="mydb"):
        """
        初始化資料庫連線

        Args:
            host: 資料庫主機
            port: 連接埠
            database: 資料庫名稱
        """
        self.host = host
        self.port = port
        self.database = database
        self._connection = None
        print(f"建立資料庫連線: {host}:{port}/{database}")

    def connect(self):
        """
        建立實際的資料庫連線
        """
        if self._connection is None:
            # 這裡應該是實際的連線邏輯
            self._connection = f"Connection to {self.database}"
            print("連線成功建立")
        return self._connection

    def execute(self, query):
        """
        執行查詢

        Args:
            query: SQL 查詢字串
        """
        if self._connection is None:
            self.connect()
        print(f"執行查詢: {query}")
        return f"Result of: {query}"

# 多執行緒測試
def test_thread_safety():
    """
    測試 Singleton 在多執行緒環境下的行為
    """
    instances = []

    def create_instance():
        """
        建立實例並加入列表
        """
        instance = DatabaseConnection()
        instances.append(instance)

    # 建立多個執行緒同時嘗試取得實例
    threads = []
    for i in range(10):
        thread = threading.Thread(target=create_instance)
        threads.append(thread)
        thread.start()

    # 等待所有執行緒完成
    for thread in threads:
        thread.join()

    # 驗證所有執行緒取得的都是同一個實例
    first_instance = instances[0]
    all_same = all(instance is first_instance for instance in instances)
    print(f"所有執行緒取得相同實例: {all_same}")
    print(f"總共取得 {len(instances)} 個參考,實際實例數: {len(set(id(i) for i in instances))}")

if __name__ == "__main__":
    test_thread_safety()

Factory 模式家族

Factory 模式是建立型模式中最常用的一類,它將物件的建立邏輯封裝起來,讓客戶端程式碼不需要知道具體要建立哪個類別的實例。這類模式包括 Simple Factory、Factory Method 和 Abstract Factory,它們的複雜度和靈活性依序遞增。

from abc import ABC, abstractmethod
from typing import Dict, Type

class Document(ABC):
    """
    文件抽象基礎類別

    定義所有文件類型必須實作的介面
    使用 ABC(Abstract Base Class)確保子類別必須實作特定方法
    """

    @abstractmethod
    def create(self):
        """
        建立文件的抽象方法

        每種文件類型必須實作自己的建立邏輯
        """
        pass

    @abstractmethod
    def save(self, filename: str):
        """
        儲存文件的抽象方法

        Args:
            filename: 檔案名稱
        """
        pass

    @abstractmethod
    def get_type(self) -> str:
        """
        取得文件類型

        Returns:
            文件類型字串
        """
        pass

class PDFDocument(Document):
    """
    PDF 文件類別

    實作 PDF 格式文件的建立和儲存邏輯
    """

    def __init__(self):
        """
        初始化 PDF 文件
        """
        self._content = []
        self._metadata = {}

    def create(self):
        """
        建立 PDF 文件結構

        初始化 PDF 特有的結構,如頁面、字型等
        """
        print("建立 PDF 文件結構...")
        self._metadata['format'] = 'PDF'
        self._metadata['version'] = '1.7'
        # 實際應用中這裡會初始化 PDF 結構
        return self

    def save(self, filename: str):
        """
        儲存為 PDF 檔案

        Args:
            filename: 檔案名稱(不含副檔名)
        """
        full_filename = f"{filename}.pdf"
        print(f"儲存 PDF 文件: {full_filename}")
        # 實際應用中這裡會寫入檔案

    def get_type(self) -> str:
        """
        傳回文件類型
        """
        return "PDF"

    def add_page(self, content: str):
        """
        新增頁面(PDF 特有方法)

        Args:
            content: 頁面內容
        """
        self._content.append(content)
        print(f"新增 PDF 頁面,目前共 {len(self._content)} 頁")

class WordDocument(Document):
    """
    Word 文件類別

    實作 Word 格式文件的建立和儲存邏輯
    """

    def __init__(self):
        """
        初始化 Word 文件
        """
        self._paragraphs = []
        self._styles = {}

    def create(self):
        """
        建立 Word 文件結構

        初始化 Word 特有的結構,如段落、樣式等
        """
        print("建立 Word 文件結構...")
        self._styles['default'] = {'font': 'Calibri', 'size': 11}
        return self

    def save(self, filename: str):
        """
        儲存為 Word 檔案

        Args:
            filename: 檔案名稱(不含副檔名)
        """
        full_filename = f"{filename}.docx"
        print(f"儲存 Word 文件: {full_filename}")

    def get_type(self) -> str:
        """
        傳回文件類型
        """
        return "Word"

    def add_paragraph(self, text: str, style: str = 'default'):
        """
        新增段落(Word 特有方法)

        Args:
            text: 段落文字
            style: 套用的樣式名稱
        """
        self._paragraphs.append({'text': text, 'style': style})
        print(f"新增段落,目前共 {len(self._paragraphs)} 個段落")

class ExcelDocument(Document):
    """
    Excel 文件類別

    實作 Excel 格式文件的建立和儲存邏輯
    """

    def __init__(self):
        """
        初始化 Excel 文件
        """
        self._sheets = {}
        self._active_sheet = None

    def create(self):
        """
        建立 Excel 文件結構

        初始化預設的工作表
        """
        print("建立 Excel 文件結構...")
        self._sheets['Sheet1'] = []
        self._active_sheet = 'Sheet1'
        return self

    def save(self, filename: str):
        """
        儲存為 Excel 檔案

        Args:
            filename: 檔案名稱(不含副檔名)
        """
        full_filename = f"{filename}.xlsx"
        print(f"儲存 Excel 文件: {full_filename}")

    def get_type(self) -> str:
        """
        傳回文件類型
        """
        return "Excel"

    def add_sheet(self, name: str):
        """
        新增工作表(Excel 特有方法)

        Args:
            name: 工作表名稱
        """
        self._sheets[name] = []
        print(f"新增工作表: {name}")

class DocumentFactory:
    """
    文件工廠類別(Simple Factory 模式)

    封裝文件建立邏輯,客戶端不需要知道具體類別
    可以輕鬆新增新的文件類型而不影響客戶端程式碼
    """

    # 註冊的文件類型對應表
    # 使用字典讓新增類型更加方便
    _document_types: Dict[str, Type[Document]] = {
        'pdf': PDFDocument,
        'word': WordDocument,
        'excel': ExcelDocument
    }

    @classmethod
    def register_type(cls, doc_type: str, doc_class: Type[Document]):
        """
        註冊新的文件類型

        這個方法讓工廠具有可擴充性
        可以在執行時期動態新增新的文件類型

        Args:
            doc_type: 文件類型識別碼
            doc_class: 文件類別
        """
        cls._document_types[doc_type.lower()] = doc_class
        print(f"已註冊文件類型: {doc_type}")

    @classmethod
    def create_document(cls, doc_type: str) -> Document:
        """
        建立文件的工廠方法

        根據指定的類型建立對應的文件實例

        Args:
            doc_type: 文件類型(pdf, word, excel 等)

        Returns:
            建立並初始化的文件實例

        Raises:
            ValueError: 當指定的類型不存在時
        """
        # 轉換為小寫以統一處理
        doc_type_lower = doc_type.lower()

        # 檢查類型是否存在
        if doc_type_lower not in cls._document_types:
            available = ', '.join(cls._document_types.keys())
            raise ValueError(
                f"不支援的文件類型: {doc_type}。"
                f"可用類型: {available}"
            )

        # 取得對應的類別並建立實例
        document_class = cls._document_types[doc_type_lower]
        document = document_class()

        # 呼叫建立方法進行初始化
        document.create()

        return document

    @classmethod
    def get_available_types(cls) -> list:
        """
        取得所有可用的文件類型

        Returns:
            文件類型列表
        """
        return list(cls._document_types.keys())

# 使用範例
if __name__ == "__main__":
    # 使用工廠建立不同類型的文件
    print("=== 使用 DocumentFactory 建立文件 ===\n")

    # 建立 PDF 文件
    pdf_doc = DocumentFactory.create_document('pdf')
    pdf_doc.add_page("第一頁內容")
    pdf_doc.save("report")

    print()

    # 建立 Word 文件
    word_doc = DocumentFactory.create_document('word')
    word_doc.add_paragraph("這是第一段")
    word_doc.save("letter")

    print()

    # 建立 Excel 文件
    excel_doc = DocumentFactory.create_document('excel')
    excel_doc.add_sheet("數據分析")
    excel_doc.save("data")

    print()

    # 顯示可用類型
    print(f"可用文件類型: {DocumentFactory.get_available_types()}")

Abstract Factory 模式

當系統需要建立一系列相關的物件時,Abstract Factory 模式提供了更高層次的抽象。這個模式特別適合用於需要支援多種主題、平台或風格的應用程式:

from abc import ABC, abstractmethod

class Button(ABC):
    """
    按鈕抽象類別

    定義所有按鈕必須實作的介面
    """

    @abstractmethod
    def render(self):
        """
        渲染按鈕
        """
        pass

    @abstractmethod
    def on_click(self, callback):
        """
        設定點擊事件處理器

        Args:
            callback: 點擊時要執行的函式
        """
        pass

class Checkbox(ABC):
    """
    核取方塊抽象類別

    定義所有核取方塊必須實作的介面
    """

    @abstractmethod
    def render(self):
        """
        渲染核取方塊
        """
        pass

    @abstractmethod
    def toggle(self):
        """
        切換選取狀態
        """
        pass

class TextField(ABC):
    """
    文字欄位抽象類別

    定義所有文字欄位必須實作的介面
    """

    @abstractmethod
    def render(self):
        """
        渲染文字欄位
        """
        pass

    @abstractmethod
    def get_value(self) -> str:
        """
        取得欄位值

        Returns:
            欄位中的文字
        """
        pass

# Windows 風格的 UI 元件
class WindowsButton(Button):
    """
    Windows 風格按鈕

    實作 Windows 平台特有的按鈕外觀和行為
    """

    def __init__(self, label: str):
        """
        初始化 Windows 按鈕

        Args:
            label: 按鈕文字
        """
        self.label = label
        self._callback = None

    def render(self):
        """
        渲染 Windows 風格的按鈕

        使用 Windows 特有的視覺樣式
        """
        print(f"[Windows Button: {self.label}]")

    def on_click(self, callback):
        """
        設定點擊處理器

        Args:
            callback: 回呼函式
        """
        self._callback = callback
        print(f"已為 Windows 按鈕設定點擊事件")

class WindowsCheckbox(Checkbox):
    """
    Windows 風格核取方塊
    """

    def __init__(self, label: str):
        """
        初始化

        Args:
            label: 標籤文字
        """
        self.label = label
        self._checked = False

    def render(self):
        """
        渲染 Windows 風格的核取方塊
        """
        status = "☑" if self._checked else "☐"
        print(f"{status} Windows Checkbox: {self.label}")

    def toggle(self):
        """
        切換選取狀態
        """
        self._checked = not self._checked
        print(f"Windows 核取方塊已切換為: {'選取' if self._checked else '未選取'}")

class WindowsTextField(TextField):
    """
    Windows 風格文字欄位
    """

    def __init__(self, placeholder: str = ""):
        """
        初始化

        Args:
            placeholder: 佔位文字
        """
        self.placeholder = placeholder
        self._value = ""

    def render(self):
        """
        渲染 Windows 風格的文字欄位
        """
        display = self._value if self._value else self.placeholder
        print(f"|__{display}__|  (Windows TextField)")

    def get_value(self) -> str:
        """
        取得欄位值

        Returns:
            目前的文字值
        """
        return self._value

# macOS 風格的 UI 元件
class MacButton(Button):
    """
    macOS 風格按鈕

    實作 macOS 平台特有的按鈕外觀和行為
    """

    def __init__(self, label: str):
        """
        初始化 macOS 按鈕

        Args:
            label: 按鈕文字
        """
        self.label = label
        self._callback = None

    def render(self):
        """
        渲染 macOS 風格的按鈕

        使用 macOS 特有的圓角視覺樣式
        """
        print(f"( macOS Button: {self.label} )")

    def on_click(self, callback):
        """
        設定點擊處理器
        """
        self._callback = callback
        print(f"已為 macOS 按鈕設定點擊事件")

class MacCheckbox(Checkbox):
    """
    macOS 風格核取方塊
    """

    def __init__(self, label: str):
        self.label = label
        self._checked = False

    def render(self):
        """
        渲染 macOS 風格的核取方塊
        """
        status = "✓" if self._checked else "○"
        print(f"{status} macOS Checkbox: {self.label}")

    def toggle(self):
        self._checked = not self._checked
        print(f"macOS 核取方塊已切換為: {'選取' if self._checked else '未選取'}")

class MacTextField(TextField):
    """
    macOS 風格文字欄位
    """

    def __init__(self, placeholder: str = ""):
        self.placeholder = placeholder
        self._value = ""

    def render(self):
        display = self._value if self._value else self.placeholder
        print(f"[ {display} ]  (macOS TextField)")

    def get_value(self) -> str:
        return self._value

class GUIFactory(ABC):
    """
    GUI 工廠抽象類別

    定義建立一系列相關 UI 元件的介面
    每個具體工廠將建立特定平台風格的元件
    """

    @abstractmethod
    def create_button(self, label: str) -> Button:
        """
        建立按鈕

        Args:
            label: 按鈕文字

        Returns:
            按鈕實例
        """
        pass

    @abstractmethod
    def create_checkbox(self, label: str) -> Checkbox:
        """
        建立核取方塊

        Args:
            label: 標籤文字

        Returns:
            核取方塊實例
        """
        pass

    @abstractmethod
    def create_text_field(self, placeholder: str = "") -> TextField:
        """
        建立文字欄位

        Args:
            placeholder: 佔位文字

        Returns:
            文字欄位實例
        """
        pass

class WindowsFactory(GUIFactory):
    """
    Windows UI 工廠

    建立 Windows 風格的 UI 元件
    """

    def create_button(self, label: str) -> Button:
        """
        建立 Windows 按鈕
        """
        return WindowsButton(label)

    def create_checkbox(self, label: str) -> Checkbox:
        """
        建立 Windows 核取方塊
        """
        return WindowsCheckbox(label)

    def create_text_field(self, placeholder: str = "") -> TextField:
        """
        建立 Windows 文字欄位
        """
        return WindowsTextField(placeholder)

class MacFactory(GUIFactory):
    """
    macOS UI 工廠

    建立 macOS 風格的 UI 元件
    """

    def create_button(self, label: str) -> Button:
        """
        建立 macOS 按鈕
        """
        return MacButton(label)

    def create_checkbox(self, label: str) -> Checkbox:
        """
        建立 macOS 核取方塊
        """
        return MacCheckbox(label)

    def create_text_field(self, placeholder: str = "") -> TextField:
        """
        建立 macOS 文字欄位
        """
        return MacTextField(placeholder)

class Application:
    """
    應用程式類別

    使用 Abstract Factory 來建立 UI 元件
    不需要知道具體的平台實作細節
    """

    def __init__(self, factory: GUIFactory):
        """
        初始化應用程式

        Args:
            factory: UI 工廠實例
        """
        self.factory = factory
        self.button = None
        self.checkbox = None
        self.text_field = None

    def create_ui(self):
        """
        使用工廠建立 UI 元件

        透過抽象工廠介面建立元件
        實際的元件類型由具體工廠決定
        """
        # 使用工廠建立元件
        self.button = self.factory.create_button("送出")
        self.checkbox = self.factory.create_checkbox("記住我")
        self.text_field = self.factory.create_text_field("請輸入使用者名稱")

    def render(self):
        """
        渲染所有 UI 元件
        """
        print("\n=== 應用程式 UI ===")
        self.text_field.render()
        self.checkbox.render()
        self.button.render()
        print("==================\n")

def get_factory_for_platform(platform: str) -> GUIFactory:
    """
    根據平台取得對應的工廠

    這是一個簡單的工廠選擇邏輯
    實際應用中可能會根據系統偵測自動選擇

    Args:
        platform: 平台名稱

    Returns:
        對應平台的 GUI 工廠
    """
    factories = {
        'windows': WindowsFactory,
        'macos': MacFactory
    }

    factory_class = factories.get(platform.lower())
    if factory_class is None:
        raise ValueError(f"不支援的平台: {platform}")

    return factory_class()

# 使用範例
if __name__ == "__main__":
    # 建立 Windows 風格的應用程式
    windows_factory = get_factory_for_platform('windows')
    windows_app = Application(windows_factory)
    windows_app.create_ui()
    windows_app.render()

    # 建立 macOS 風格的應用程式
    mac_factory = get_factory_for_platform('macos')
    mac_app = Application(mac_factory)
    mac_app.create_ui()
    mac_app.render()

Builder 模式

Builder 模式用於建立複雜物件,將物件的建構過程與其表示分離。當一個物件有許多可選參數,或者建構過程包含多個步驟時,Builder 模式特別有用:

from typing import Optional, List
from datetime import datetime

class EmailMessage:
    """
    電子郵件訊息類別

    這是 Builder 模式要建構的複雜物件
    包含許多可選的屬性
    """

    def __init__(self):
        """
        初始化電子郵件

        所有屬性都設為預設值
        實際的值由 Builder 設定
        """
        self.sender: str = ""
        self.recipients: List[str] = []
        self.cc: List[str] = []
        self.bcc: List[str] = []
        self.subject: str = ""
        self.body: str = ""
        self.html_body: Optional[str] = None
        self.attachments: List[dict] = []
        self.priority: str = "normal"
        self.timestamp: datetime = datetime.now()
        self.headers: dict = {}

    def __str__(self):
        """
        格式化顯示電子郵件資訊
        """
        result = [
            f"寄件者: {self.sender}",
            f"收件者: {', '.join(self.recipients)}",
        ]

        if self.cc:
            result.append(f"副本: {', '.join(self.cc)}")
        if self.bcc:
            result.append(f"密件副本: {', '.join(self.bcc)}")

        result.extend([
            f"主旨: {self.subject}",
            f"優先順序: {self.priority}",
            f"時間: {self.timestamp}",
            f"附件數量: {len(self.attachments)}",
            "---",
            self.body
        ])

        return "\n".join(result)

class EmailBuilder:
    """
    電子郵件建構器

    提供流暢的 API 來建構 EmailMessage 物件
    使用 Method Chaining 讓程式碼更易讀
    """

    def __init__(self):
        """
        初始化建構器

        建立新的 EmailMessage 實例
        """
        self._email = EmailMessage()

    def sender(self, email_address: str) -> 'EmailBuilder':
        """
        設定寄件者

        Args:
            email_address: 寄件者電子郵件地址

        Returns:
            建構器本身,支援 Method Chaining
        """
        # 驗證電子郵件格式(簡化版)
        if '@' not in email_address:
            raise ValueError(f"無效的電子郵件地址: {email_address}")

        self._email.sender = email_address
        return self

    def to(self, *recipients: str) -> 'EmailBuilder':
        """
        新增收件者

        Args:
            *recipients: 一個或多個收件者電子郵件地址

        Returns:
            建構器本身
        """
        for recipient in recipients:
            if '@' not in recipient:
                raise ValueError(f"無效的電子郵件地址: {recipient}")
            self._email.recipients.append(recipient)
        return self

    def cc(self, *recipients: str) -> 'EmailBuilder':
        """
        新增副本收件者

        Args:
            *recipients: 副本收件者地址

        Returns:
            建構器本身
        """
        for recipient in recipients:
            self._email.cc.append(recipient)
        return self

    def bcc(self, *recipients: str) -> 'EmailBuilder':
        """
        新增密件副本收件者

        Args:
            *recipients: 密件副本收件者地址

        Returns:
            建構器本身
        """
        for recipient in recipients:
            self._email.bcc.append(recipient)
        return self

    def subject(self, subject: str) -> 'EmailBuilder':
        """
        設定郵件主旨

        Args:
            subject: 主旨文字

        Returns:
            建構器本身
        """
        self._email.subject = subject
        return self

    def body(self, content: str) -> 'EmailBuilder':
        """
        設定純文字內容

        Args:
            content: 郵件內容

        Returns:
            建構器本身
        """
        self._email.body = content
        return self

    def html_body(self, html_content: str) -> 'EmailBuilder':
        """
        設定 HTML 內容

        Args:
            html_content: HTML 格式的郵件內容

        Returns:
            建構器本身
        """
        self._email.html_body = html_content
        return self

    def attach(self, filename: str, content: bytes,
               mime_type: str = "application/octet-stream") -> 'EmailBuilder':
        """
        新增附件

        Args:
            filename: 附件檔名
            content: 附件內容(二進位)
            mime_type: MIME 類型

        Returns:
            建構器本身
        """
        attachment = {
            'filename': filename,
            'content': content,
            'mime_type': mime_type
        }
        self._email.attachments.append(attachment)
        return self

    def priority(self, level: str) -> 'EmailBuilder':
        """
        設定優先順序

        Args:
            level: 優先順序(low, normal, high)

        Returns:
            建構器本身
        """
        valid_levels = ['low', 'normal', 'high']
        if level.lower() not in valid_levels:
            raise ValueError(f"無效的優先順序: {level}。有效值: {valid_levels}")

        self._email.priority = level.lower()
        return self

    def add_header(self, name: str, value: str) -> 'EmailBuilder':
        """
        新增自訂標頭

        Args:
            name: 標頭名稱
            value: 標頭值

        Returns:
            建構器本身
        """
        self._email.headers[name] = value
        return self

    def build(self) -> EmailMessage:
        """
        建構並驗證電子郵件

        在傳回物件之前進行最終驗證
        確保必要的欄位都已設定

        Returns:
            建構完成的 EmailMessage 實例

        Raises:
            ValueError: 當必要欄位未設定時
        """
        # 驗證必要欄位
        if not self._email.sender:
            raise ValueError("必須設定寄件者")

        if not self._email.recipients:
            raise ValueError("必須至少有一個收件者")

        if not self._email.subject:
            raise ValueError("必須設定主旨")

        if not self._email.body and not self._email.html_body:
            raise ValueError("必須設定郵件內容")

        # 傳回建構好的物件
        return self._email

class EmailDirector:
    """
    電子郵件導演類別

    封裝常用的電子郵件建構流程
    提供預定義的建構方法
    """

    @staticmethod
    def create_simple_email(builder: EmailBuilder,
                           sender: str,
                           recipient: str,
                           subject: str,
                           body: str) -> EmailMessage:
        """
        建立簡單的電子郵件

        Args:
            builder: EmailBuilder 實例
            sender: 寄件者
            recipient: 收件者
            subject: 主旨
            body: 內容

        Returns:
            建構完成的電子郵件
        """
        return (builder
                .sender(sender)
                .to(recipient)
                .subject(subject)
                .body(body)
                .build())

    @staticmethod
    def create_newsletter(builder: EmailBuilder,
                         sender: str,
                         recipients: List[str],
                         subject: str,
                         html_content: str) -> EmailMessage:
        """
        建立電子報

        電子報通常是 HTML 格式且有多個收件者

        Args:
            builder: EmailBuilder 實例
            sender: 寄件者
            recipients: 收件者列表
            subject: 主旨
            html_content: HTML 內容

        Returns:
            建構完成的電子郵件
        """
        return (builder
                .sender(sender)
                .to(*recipients)
                .subject(subject)
                .body("請使用支援 HTML 的郵件客戶端檢視此郵件。")
                .html_body(html_content)
                .add_header("List-Unsubscribe", "<mailto:[email protected]>")
                .build())

# 使用範例
if __name__ == "__main__":
    # 使用 Builder 建構複雜的電子郵件
    print("=== 使用 Builder 建構電子郵件 ===\n")

    # 建構一封包含附件的高優先順序郵件
    email = (EmailBuilder()
             .sender("[email protected]")
             .to("[email protected]", "[email protected]")
             .cc("[email protected]")
             .subject("專案進度報告")
             .body("請查閱附件中的專案進度報告。\n\n祝好")
             .attach("report.pdf", b"PDF content here", "application/pdf")
             .priority("high")
             .add_header("X-Project-ID", "PRJ-2025-001")
             .build())

    print(email)
    print("\n" + "=" * 50 + "\n")

    # 使用 Director 建立電子報
    newsletter = EmailDirector.create_newsletter(
        EmailBuilder(),
        "[email protected]",
        ["[email protected]", "[email protected]", "[email protected]"],
        "2025年11月份電子報",
        "<h1>本月精選</h1><p>歡迎閱讀本月電子報...</p>"
    )

    print(newsletter)

結構型模式詳解

結構型模式關注類別和物件的組合方式,目的是讓系統結構更加靈活、更容易擴充。這類模式透過定義物件之間的關係,使系統能夠在不修改現有程式碼的情況下調整和擴充功能。

Adapter 模式

Adapter 模式將一個類別的介面轉換成客戶端期望的另一個介面,讓原本因介面不相容而無法協同工作的類別能夠一起運作。這個模式在整合第三方程式庫或舊系統時特別有用:

from abc import ABC, abstractmethod
from typing import Dict, Any
import json
import xml.etree.ElementTree as ET

class DataProcessor(ABC):
    """
    資料處理器抽象介面

    定義系統期望的統一資料處理介面
    所有資料來源都應該符合這個介面
    """

    @abstractmethod
    def get_data(self) -> Dict[str, Any]:
        """
        取得資料

        Returns:
            字典格式的資料
        """
        pass

    @abstractmethod
    def process(self) -> str:
        """
        處理資料並傳回結果

        Returns:
            處理結果字串
        """
        pass

class ModernAPIClient:
    """
    現代 API 客戶端

    已經符合系統期望的介面
    """

    def __init__(self, api_url: str):
        """
        初始化 API 客戶端

        Args:
            api_url: API 端點 URL
        """
        self.api_url = api_url
        self._data = None

    def fetch(self) -> Dict[str, Any]:
        """
        從 API 取得資料

        Returns:
            JSON 格式的資料
        """
        # 模擬 API 回應
        self._data = {
            "users": [
                {"id": 1, "name": "Alice", "email": "[email protected]"},
                {"id": 2, "name": "Bob", "email": "[email protected]"}
            ],
            "total": 2
        }
        return self._data

    def get_data(self) -> Dict[str, Any]:
        """
        取得快取的資料

        Returns:
            資料字典
        """
        if self._data is None:
            self.fetch()
        return self._data

    def process(self) -> str:
        """
        處理資料

        Returns:
            處理結果
        """
        data = self.get_data()
        return f"處理了 {data['total']} 筆使用者資料"

class LegacyXMLService:
    """
    舊版 XML 服務

    這是一個舊系統,使用 XML 格式和不同的方法名稱
    無法直接與現代系統介面相容
    """

    def __init__(self, service_endpoint: str):
        """
        初始化 XML 服務

        Args:
            service_endpoint: 服務端點
        """
        self.endpoint = service_endpoint

    def retrieve_xml_data(self) -> str:
        """
        取得 XML 格式的資料

        這是舊系統的方法,傳回 XML 字串

        Returns:
            XML 格式的資料字串
        """
        # 模擬舊系統回傳的 XML 資料
        xml_data = """<?xml version="1.0"?>
        <response>
            <users>
                <user>
                    <id>1</id>
                    <name>Charlie</name>
                    <email>[email protected]</email>
                </user>
                <user>
                    <id>2</id>
                    <name>Diana</name>
                    <email>[email protected]</email>
                </user>
            </users>
            <count>2</count>
        </response>
        """
        return xml_data

    def execute_legacy_process(self, xml_data: str) -> str:
        """
        舊系統的處理方法

        Args:
            xml_data: XML 資料

        Returns:
            處理結果
        """
        return f"舊系統處理完成,端點: {self.endpoint}"

class LegacyXMLAdapter(DataProcessor):
    """
    舊版 XML 服務的適配器

    將 LegacyXMLService 的介面轉換為 DataProcessor 介面
    這樣舊系統就能與新系統一起使用了
    """

    def __init__(self, legacy_service: LegacyXMLService):
        """
        初始化適配器

        Args:
            legacy_service: 舊版 XML 服務實例
        """
        self._legacy_service = legacy_service
        self._cached_data = None

    def _parse_xml_to_dict(self, xml_string: str) -> Dict[str, Any]:
        """
        將 XML 轉換為字典格式

        這是適配器的核心轉換邏輯

        Args:
            xml_string: XML 字串

        Returns:
            轉換後的字典
        """
        # 解析 XML
        root = ET.fromstring(xml_string)

        # 轉換為字典格式
        users = []
        for user_elem in root.findall('.//user'):
            user = {
                'id': int(user_elem.find('id').text),
                'name': user_elem.find('name').text,
                'email': user_elem.find('email').text
            }
            users.append(user)

        count_elem = root.find('.//count')
        count = int(count_elem.text) if count_elem is not None else len(users)

        return {
            'users': users,
            'total': count
        }

    def get_data(self) -> Dict[str, Any]:
        """
        實作 DataProcessor 介面的 get_data 方法

        將舊系統的 XML 資料轉換為字典格式

        Returns:
            字典格式的資料
        """
        if self._cached_data is None:
            # 從舊系統取得 XML 資料
            xml_data = self._legacy_service.retrieve_xml_data()
            # 轉換為標準格式
            self._cached_data = self._parse_xml_to_dict(xml_data)

        return self._cached_data

    def process(self) -> str:
        """
        實作 DataProcessor 介面的 process 方法

        Returns:
            處理結果字串
        """
        data = self.get_data()
        # 呼叫舊系統的處理方法
        xml_data = self._legacy_service.retrieve_xml_data()
        legacy_result = self._legacy_service.execute_legacy_process(xml_data)

        return f"透過適配器處理了 {data['total']} 筆資料。{legacy_result}"

class CSVDataSource:
    """
    CSV 資料來源

    另一個需要適配的資料來源
    使用完全不同的方法和資料格式
    """

    def __init__(self, file_path: str):
        """
        初始化 CSV 資料來源

        Args:
            file_path: CSV 檔案路徑
        """
        self.file_path = file_path

    def read_csv(self) -> list:
        """
        讀取 CSV 資料

        Returns:
            CSV 資料列表
        """
        # 模擬 CSV 資料
        return [
            ["id", "name", "email"],
            ["1", "Eve", "[email protected]"],
            ["2", "Frank", "[email protected]"],
            ["3", "Grace", "[email protected]"]
        ]

    def get_row_count(self) -> int:
        """
        取得資料列數

        Returns:
            列數(不含標題列)
        """
        return len(self.read_csv()) - 1

class CSVAdapter(DataProcessor):
    """
    CSV 資料來源的適配器

    將 CSV 資料轉換為標準的字典格式
    """

    def __init__(self, csv_source: CSVDataSource):
        """
        初始化適配器

        Args:
            csv_source: CSV 資料來源實例
        """
        self._csv_source = csv_source
        self._cached_data = None

    def get_data(self) -> Dict[str, Any]:
        """
        將 CSV 資料轉換為字典格式

        Returns:
            標準格式的資料字典
        """
        if self._cached_data is None:
            rows = self._csv_source.read_csv()
            headers = rows[0]  # 第一列是標題

            users = []
            for row in rows[1:]:  # 跳過標題列
                user = {
                    headers[i]: int(row[i]) if headers[i] == 'id' else row[i]
                    for i in range(len(headers))
                }
                users.append(user)

            self._cached_data = {
                'users': users,
                'total': len(users)
            }

        return self._cached_data

    def process(self) -> str:
        """
        處理 CSV 資料

        Returns:
            處理結果
        """
        data = self.get_data()
        return f"處理了來自 {self._csv_source.file_path}{data['total']} 筆資料"

class DataAggregator:
    """
    資料聚合器

    使用統一的 DataProcessor 介面處理來自不同來源的資料
    這展示了 Adapter 模式的價值:客戶端程式碼不需要知道資料來源的具體實作
    """

    def __init__(self):
        """
        初始化聚合器
        """
        self._processors: list = []

    def add_processor(self, processor: DataProcessor):
        """
        新增資料處理器

        Args:
            processor: 符合 DataProcessor 介面的處理器
        """
        self._processors.append(processor)

    def aggregate_all(self) -> Dict[str, Any]:
        """
        聚合所有資料來源的資料

        Returns:
            聚合後的資料
        """
        all_users = []
        total = 0

        for processor in self._processors:
            data = processor.get_data()
            all_users.extend(data['users'])
            total += data['total']

        return {
            'users': all_users,
            'total': total,
            'sources': len(self._processors)
        }

    def process_all(self) -> list:
        """
        處理所有資料來源

        Returns:
            各處理器的結果列表
        """
        results = []
        for processor in self._processors:
            result = processor.process()
            results.append(result)
        return results

# 使用範例
if __name__ == "__main__":
    print("=== Adapter 模式示範 ===\n")

    # 建立資料聚合器
    aggregator = DataAggregator()

    # 新增現代 API 客戶端(已符合介面)
    modern_api = ModernAPIClient("https://api.example.com")
    aggregator.add_processor(modern_api)

    # 新增舊版 XML 服務(透過適配器)
    legacy_service = LegacyXMLService("http://legacy.example.com/xml")
    xml_adapter = LegacyXMLAdapter(legacy_service)
    aggregator.add_processor(xml_adapter)

    # 新增 CSV 資料來源(透過適配器)
    csv_source = CSVDataSource("data/users.csv")
    csv_adapter = CSVAdapter(csv_source)
    aggregator.add_processor(csv_adapter)

    # 聚合所有資料
    aggregated = aggregator.aggregate_all()
    print(f"聚合資料來源數: {aggregated['sources']}")
    print(f"總使用者數: {aggregated['total']}")
    print(f"使用者列表: {json.dumps(aggregated['users'], indent=2, ensure_ascii=False)}")

    print("\n--- 處理結果 ---")
    for result in aggregator.process_all():
        print(f"  {result}")

Decorator 模式

Decorator 模式動態地為物件添加額外的職責。與繼承相比,Decorator 模式提供了更加靈活的方式來擴充功能。Python 的裝飾器語法讓這個模式的實作特別優雅:

from abc import ABC, abstractmethod
from typing import Optional
import time
import functools

class DataService(ABC):
    """
    資料服務抽象介面

    定義資料服務的基本操作
    """

    @abstractmethod
    def get_data(self, key: str) -> Optional[str]:
        """
        根據鍵取得資料

        Args:
            key: 資料鍵

        Returns:
            資料值,若不存在則傳回 None
        """
        pass

    @abstractmethod
    def set_data(self, key: str, value: str) -> bool:
        """
        設定資料

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        pass

class BasicDataService(DataService):
    """
    基礎資料服務

    提供簡單的鍵值對儲存功能
    """

    def __init__(self):
        """
        初始化基礎資料服務
        """
        self._storage = {}

    def get_data(self, key: str) -> Optional[str]:
        """
        取得資料

        Args:
            key: 資料鍵

        Returns:
            資料值
        """
        # 模擬緩慢的資料存取
        time.sleep(0.1)
        return self._storage.get(key)

    def set_data(self, key: str, value: str) -> bool:
        """
        設定資料

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        time.sleep(0.1)
        self._storage[key] = value
        return True

class DataServiceDecorator(DataService):
    """
    資料服務裝飾器基礎類別

    所有裝飾器都繼承自這個類別
    實作基本的委派邏輯
    """

    def __init__(self, service: DataService):
        """
        初始化裝飾器

        Args:
            service: 被裝飾的服務
        """
        self._wrapped_service = service

    def get_data(self, key: str) -> Optional[str]:
        """
        委派給被裝飾的服務

        Args:
            key: 資料鍵

        Returns:
            資料值
        """
        return self._wrapped_service.get_data(key)

    def set_data(self, key: str, value: str) -> bool:
        """
        委派給被裝飾的服務

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        return self._wrapped_service.set_data(key, value)

class CachingDecorator(DataServiceDecorator):
    """
    快取裝飾器

    為資料服務新增快取功能
    避免重複存取緩慢的底層儲存
    """

    def __init__(self, service: DataService, cache_size: int = 100):
        """
        初始化快取裝飾器

        Args:
            service: 被裝飾的服務
            cache_size: 快取大小上限
        """
        super().__init__(service)
        self._cache = {}
        self._cache_size = cache_size
        self._cache_hits = 0
        self._cache_misses = 0

    def get_data(self, key: str) -> Optional[str]:
        """
        帶快取的資料取得

        先檢查快取,若快取未命中則從底層服務取得

        Args:
            key: 資料鍵

        Returns:
            資料值
        """
        # 檢查快取
        if key in self._cache:
            self._cache_hits += 1
            print(f"[快取命中] key={key}")
            return self._cache[key]

        # 快取未命中,從底層服務取得
        self._cache_misses += 1
        print(f"[快取未命中] key={key}")

        value = self._wrapped_service.get_data(key)

        # 儲存到快取
        if value is not None:
            self._add_to_cache(key, value)

        return value

    def set_data(self, key: str, value: str) -> bool:
        """
        設定資料並更新快取

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        result = self._wrapped_service.set_data(key, value)

        if result:
            # 更新快取
            self._add_to_cache(key, value)

        return result

    def _add_to_cache(self, key: str, value: str):
        """
        新增到快取

        如果快取已滿,移除最舊的項目

        Args:
            key: 資料鍵
            value: 資料值
        """
        if len(self._cache) >= self._cache_size:
            # 移除第一個項目(簡單的 FIFO 策略)
            oldest_key = next(iter(self._cache))
            del self._cache[oldest_key]

        self._cache[key] = value

    def get_cache_stats(self) -> dict:
        """
        取得快取統計資訊

        Returns:
            統計資訊字典
        """
        total = self._cache_hits + self._cache_misses
        hit_rate = self._cache_hits / total if total > 0 else 0

        return {
            'hits': self._cache_hits,
            'misses': self._cache_misses,
            'hit_rate': f"{hit_rate:.2%}",
            'cache_size': len(self._cache)
        }

class LoggingDecorator(DataServiceDecorator):
    """
    日誌裝飾器

    記錄所有資料存取操作
    """

    def __init__(self, service: DataService, log_prefix: str = "DataService"):
        """
        初始化日誌裝飾器

        Args:
            service: 被裝飾的服務
            log_prefix: 日誌前綴
        """
        super().__init__(service)
        self._log_prefix = log_prefix
        self._operations = []

    def get_data(self, key: str) -> Optional[str]:
        """
        記錄並執行資料取得操作

        Args:
            key: 資料鍵

        Returns:
            資料值
        """
        start_time = time.time()
        result = self._wrapped_service.get_data(key)
        elapsed = time.time() - start_time

        log_entry = {
            'operation': 'GET',
            'key': key,
            'result': 'found' if result else 'not_found',
            'time_ms': elapsed * 1000
        }
        self._operations.append(log_entry)

        print(f"[{self._log_prefix}] GET {key} -> "
              f"{'found' if result else 'not_found'} "
              f"({elapsed*1000:.2f}ms)")

        return result

    def set_data(self, key: str, value: str) -> bool:
        """
        記錄並執行資料設定操作

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        start_time = time.time()
        result = self._wrapped_service.set_data(key, value)
        elapsed = time.time() - start_time

        log_entry = {
            'operation': 'SET',
            'key': key,
            'result': 'success' if result else 'failed',
            'time_ms': elapsed * 1000
        }
        self._operations.append(log_entry)

        print(f"[{self._log_prefix}] SET {key}={value} -> "
              f"{'success' if result else 'failed'} "
              f"({elapsed*1000:.2f}ms)")

        return result

    def get_operation_log(self) -> list:
        """
        取得操作日誌

        Returns:
            操作日誌列表
        """
        return self._operations.copy()

class ValidationDecorator(DataServiceDecorator):
    """
    驗證裝飾器

    對資料進行驗證,確保符合規則
    """

    def __init__(self, service: DataService,
                 max_key_length: int = 100,
                 max_value_length: int = 10000):
        """
        初始化驗證裝飾器

        Args:
            service: 被裝飾的服務
            max_key_length: 鍵最大長度
            max_value_length: 值最大長度
        """
        super().__init__(service)
        self._max_key_length = max_key_length
        self._max_value_length = max_value_length

    def _validate_key(self, key: str):
        """
        驗證鍵

        Args:
            key: 資料鍵

        Raises:
            ValueError: 當鍵無效時
        """
        if not key:
            raise ValueError("鍵不能為空")

        if len(key) > self._max_key_length:
            raise ValueError(f"鍵長度超過上限 {self._max_key_length}")

        # 檢查非法字元
        if any(char in key for char in ['/', '\\', '\0']):
            raise ValueError("鍵包含非法字元")

    def _validate_value(self, value: str):
        """
        驗證值

        Args:
            value: 資料值

        Raises:
            ValueError: 當值無效時
        """
        if len(value) > self._max_value_length:
            raise ValueError(f"值長度超過上限 {self._max_value_length}")

    def get_data(self, key: str) -> Optional[str]:
        """
        驗證鍵後取得資料

        Args:
            key: 資料鍵

        Returns:
            資料值
        """
        self._validate_key(key)
        return self._wrapped_service.get_data(key)

    def set_data(self, key: str, value: str) -> bool:
        """
        驗證鍵和值後設定資料

        Args:
            key: 資料鍵
            value: 資料值

        Returns:
            是否成功
        """
        self._validate_key(key)
        self._validate_value(value)
        return self._wrapped_service.set_data(key, value)

# 使用範例
if __name__ == "__main__":
    print("=== Decorator 模式示範 ===\n")

    # 建立基礎服務
    basic_service = BasicDataService()

    # 套用多層裝飾器
    # 順序很重要:驗證 -> 快取 -> 日誌 -> 基礎服務
    validated_service = ValidationDecorator(basic_service)
    cached_service = CachingDecorator(validated_service)
    logged_service = LoggingDecorator(cached_service)

    # 使用裝飾後的服務
    service = logged_service

    # 設定資料
    service.set_data("user:1", "Alice")
    service.set_data("user:2", "Bob")

    print()

    # 第一次取得資料(快取未命中)
    service.get_data("user:1")

    # 第二次取得資料(快取命中)
    service.get_data("user:1")

    # 取得不存在的資料
    service.get_data("user:999")

    print("\n--- 快取統計 ---")
    print(cached_service.get_cache_stats())

行為型模式精要

行為型模式關注物件之間的互動和責任分配。這類模式封裝了處理物件間通訊、控制流程和職責委派的策略,讓系統能夠以鬆散耦合的方式實現複雜的行為。

Strategy 模式

Strategy 模式定義了一系列演算法,將每個演算法封裝起來,並讓它們可以互相替換。這個模式讓演算法的變化獨立於使用它的客戶端:

from abc import ABC, abstractmethod
from typing import List, Any
import random

class SortStrategy(ABC):
    """
    排序策略抽象基礎類別

    定義排序演算法的介面
    """

    @abstractmethod
    def sort(self, data: List[Any]) -> List[Any]:
        """
        執行排序

        Args:
            data: 要排序的資料列表

        Returns:
            排序後的列表
        """
        pass

    @abstractmethod
    def get_name(self) -> str:
        """
        取得演算法名稱

        Returns:
            演算法名稱
        """
        pass

class BubbleSortStrategy(SortStrategy):
    """
    氣泡排序策略

    簡單但效率較低的排序演算法
    時間複雜度:O(n²)
    """

    def sort(self, data: List[Any]) -> List[Any]:
        """
        使用氣泡排序演算法排序

        氣泡排序透過重複比較相鄰元素並交換位置來排序
        每一輪會將最大的元素「浮」到最後

        Args:
            data: 要排序的資料

        Returns:
            排序後的資料
        """
        # 建立副本避免修改原始資料
        result = data.copy()
        n = len(result)

        # 外層迴圈控制比較輪數
        for i in range(n):
            # 內層迴圈進行相鄰元素比較
            for j in range(0, n - i - 1):
                # 如果前面的元素比後面的大,交換它們
                if result[j] > result[j + 1]:
                    result[j], result[j + 1] = result[j + 1], result[j]

        return result

    def get_name(self) -> str:
        """
        傳回演算法名稱
        """
        return "Bubble Sort"

class QuickSortStrategy(SortStrategy):
    """
    快速排序策略

    高效的分治排序演算法
    平均時間複雜度:O(n log n)
    """

    def sort(self, data: List[Any]) -> List[Any]:
        """
        使用快速排序演算法排序

        快速排序選擇一個基準點,將小於基準的放左邊,
        大於基準的放右邊,然後遞迴排序兩邊

        Args:
            data: 要排序的資料

        Returns:
            排序後的資料
        """
        # 建立副本
        result = data.copy()
        self._quick_sort(result, 0, len(result) - 1)
        return result

    def _quick_sort(self, arr: List[Any], low: int, high: int):
        """
        快速排序的遞迴實作

        Args:
            arr: 要排序的陣列
            low: 起始索引
            high: 結束索引
        """
        if low < high:
            # 取得分割點
            pivot_index = self._partition(arr, low, high)
            # 遞迴排序左半部
            self._quick_sort(arr, low, pivot_index - 1)
            # 遞迴排序右半部
            self._quick_sort(arr, pivot_index + 1, high)

    def _partition(self, arr: List[Any], low: int, high: int) -> int:
        """
        分割陣列

        選擇最後一個元素作為基準,將陣列分割成兩部分

        Args:
            arr: 要分割的陣列
            low: 起始索引
            high: 結束索引

        Returns:
            基準元素的最終位置
        """
        pivot = arr[high]  # 選擇最後一個元素作為基準
        i = low - 1  # 小於基準的元素的邊界

        for j in range(low, high):
            if arr[j] <= pivot:
                i += 1
                arr[i], arr[j] = arr[j], arr[i]

        # 將基準放到正確的位置
        arr[i + 1], arr[high] = arr[high], arr[i + 1]
        return i + 1

    def get_name(self) -> str:
        """
        傳回演算法名稱
        """
        return "Quick Sort"

class MergeSortStrategy(SortStrategy):
    """
    合併排序策略

    穩定的分治排序演算法
    時間複雜度:O(n log n)
    """

    def sort(self, data: List[Any]) -> List[Any]:
        """
        使用合併排序演算法排序

        合併排序將陣列分成兩半,遞迴排序後再合併
        這是一個穩定的排序演算法

        Args:
            data: 要排序的資料

        Returns:
            排序後的資料
        """
        if len(data) <= 1:
            return data.copy()

        # 分割
        mid = len(data) // 2
        left = self.sort(data[:mid])
        right = self.sort(data[mid:])

        # 合併
        return self._merge(left, right)

    def _merge(self, left: List[Any], right: List[Any]) -> List[Any]:
        """
        合併兩個已排序的陣列

        Args:
            left: 左半部陣列
            right: 右半部陣列

        Returns:
            合併後的排序陣列
        """
        result = []
        i = j = 0

        # 比較兩個陣列的元素,依序放入結果
        while i < len(left) and j < len(right):
            if left[i] <= right[j]:
                result.append(left[i])
                i += 1
            else:
                result.append(right[j])
                j += 1

        # 將剩餘的元素加入結果
        result.extend(left[i:])
        result.extend(right[j:])

        return result

    def get_name(self) -> str:
        """
        傳回演算法名稱
        """
        return "Merge Sort"

class SortContext:
    """
    排序上下文

    使用策略模式來選擇和執行排序演算法
    客戶端透過這個類別來使用不同的排序策略
    """

    def __init__(self, strategy: SortStrategy = None):
        """
        初始化排序上下文

        Args:
            strategy: 初始的排序策略,預設為 None
        """
        self._strategy = strategy

    def set_strategy(self, strategy: SortStrategy):
        """
        設定排序策略

        允許在執行時期動態更換排序演算法

        Args:
            strategy: 新的排序策略
        """
        self._strategy = strategy
        print(f"已切換排序策略為: {strategy.get_name()}")

    def execute_sort(self, data: List[Any]) -> List[Any]:
        """
        執行排序

        使用目前設定的策略來排序資料

        Args:
            data: 要排序的資料

        Returns:
            排序後的資料

        Raises:
            ValueError: 當未設定策略時
        """
        if self._strategy is None:
            raise ValueError("未設定排序策略")

        print(f"使用 {self._strategy.get_name()} 排序 {len(data)} 筆資料...")
        result = self._strategy.sort(data)
        print(f"排序完成")

        return result

    def auto_select_strategy(self, data_size: int):
        """
        根據資料大小自動選擇最佳策略

        小資料量使用簡單演算法,大資料量使用高效演算法

        Args:
            data_size: 資料筆數
        """
        if data_size < 10:
            # 小資料量,使用簡單的氣泡排序
            self.set_strategy(BubbleSortStrategy())
        elif data_size < 1000:
            # 中等資料量,使用快速排序
            self.set_strategy(QuickSortStrategy())
        else:
            # 大資料量,使用穩定的合併排序
            self.set_strategy(MergeSortStrategy())

# 使用範例
if __name__ == "__main__":
    print("=== Strategy 模式示範 ===\n")

    # 建立測試資料
    test_data = [64, 34, 25, 12, 22, 11, 90]
    print(f"原始資料: {test_data}\n")

    # 建立上下文
    sorter = SortContext()

    # 使用氣泡排序
    sorter.set_strategy(BubbleSortStrategy())
    result1 = sorter.execute_sort(test_data)
    print(f"結果: {result1}\n")

    # 使用快速排序
    sorter.set_strategy(QuickSortStrategy())
    result2 = sorter.execute_sort(test_data)
    print(f"結果: {result2}\n")

    # 使用合併排序
    sorter.set_strategy(MergeSortStrategy())
    result3 = sorter.execute_sort(test_data)
    print(f"結果: {result3}\n")

    # 自動選擇策略
    print("--- 自動選擇策略 ---")
    large_data = [random.randint(1, 1000) for _ in range(100)]
    sorter.auto_select_strategy(len(large_data))
    result4 = sorter.execute_sort(large_data)
    print(f"前10筆: {result4[:10]}")

Observer 模式

Observer 模式定義了物件之間的一對多依賴關係,當一個物件的狀態改變時,所有依賴它的物件都會收到通知並自動更新。這個模式是事件驅動架構的基礎:

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from datetime import datetime
import weakref

class Event:
    """
    事件類別

    封裝事件的資訊
    """

    def __init__(self, event_type: str, data: Any = None, source: str = ""):
        """
        初始化事件

        Args:
            event_type: 事件類型
            data: 事件資料
            source: 事件來源
        """
        self.event_type = event_type
        self.data = data
        self.source = source
        self.timestamp = datetime.now()

    def __str__(self):
        """
        事件的字串表示
        """
        return f"Event({self.event_type}, source={self.source}, time={self.timestamp})"

class Observer(ABC):
    """
    觀察者抽象基礎類別

    定義觀察者必須實作的介面
    """

    @abstractmethod
    def update(self, event: Event):
        """
        接收事件通知

        Args:
            event: 事件物件
        """
        pass

    @abstractmethod
    def get_name(self) -> str:
        """
        取得觀察者名稱

        Returns:
            觀察者名稱
        """
        pass

class Subject:
    """
    主題(被觀察者)類別

    管理觀察者並發送通知
    使用 weakref 避免循環參考導致記憶體洩漏
    """

    def __init__(self, name: str = "Subject"):
        """
        初始化主題

        Args:
            name: 主題名稱
        """
        self._name = name
        # 使用弱參考儲存觀察者
        # 這樣當觀察者被刪除時,不會因為這裡的參考而無法釋放
        self._observers: List[weakref.ref] = []
        # 按事件類型分組的觀察者
        self._typed_observers: Dict[str, List[weakref.ref]] = {}

    def attach(self, observer: Observer, event_types: List[str] = None):
        """
        附加觀察者

        Args:
            observer: 要附加的觀察者
            event_types: 要訂閱的事件類型列表,None 表示訂閱所有事件
        """
        # 建立弱參考
        observer_ref = weakref.ref(observer)

        if event_types is None:
            # 訂閱所有事件
            self._observers.append(observer_ref)
            print(f"觀察者 '{observer.get_name()}' 已訂閱 '{self._name}' 的所有事件")
        else:
            # 訂閱特定事件
            for event_type in event_types:
                if event_type not in self._typed_observers:
                    self._typed_observers[event_type] = []
                self._typed_observers[event_type].append(observer_ref)
            print(f"觀察者 '{observer.get_name()}' 已訂閱 '{self._name}' 的事件: {event_types}")

    def detach(self, observer: Observer):
        """
        移除觀察者

        Args:
            observer: 要移除的觀察者
        """
        # 從通用觀察者列表移除
        self._observers = [
            ref for ref in self._observers
            if ref() is not None and ref() is not observer
        ]

        # 從分類觀察者列表移除
        for event_type in self._typed_observers:
            self._typed_observers[event_type] = [
                ref for ref in self._typed_observers[event_type]
                if ref() is not None and ref() is not observer
            ]

        print(f"觀察者 '{observer.get_name()}' 已從 '{self._name}' 取消訂閱")

    def notify(self, event: Event):
        """
        通知觀察者

        Args:
            event: 要發送的事件
        """
        event.source = self._name
        notified_count = 0

        # 通知訂閱所有事件的觀察者
        for observer_ref in self._observers:
            observer = observer_ref()
            if observer is not None:
                observer.update(event)
                notified_count += 1

        # 通知訂閱特定事件的觀察者
        if event.event_type in self._typed_observers:
            for observer_ref in self._typed_observers[event.event_type]:
                observer = observer_ref()
                if observer is not None:
                    observer.update(event)
                    notified_count += 1

        print(f"事件 '{event.event_type}' 已通知 {notified_count} 個觀察者")

    def _cleanup_dead_references(self):
        """
        清理已失效的弱參考

        當觀察者被刪除後,清理對應的弱參考
        """
        self._observers = [ref for ref in self._observers if ref() is not None]

        for event_type in self._typed_observers:
            self._typed_observers[event_type] = [
                ref for ref in self._typed_observers[event_type]
                if ref() is not None
            ]

class StockMarket(Subject):
    """
    股票市場(具體主題)

    模擬股票價格變動並通知觀察者
    """

    def __init__(self):
        """
        初始化股票市場
        """
        super().__init__("StockMarket")
        self._prices: Dict[str, float] = {}

    def set_price(self, symbol: str, price: float):
        """
        設定股票價格

        當價格變動時,發送通知

        Args:
            symbol: 股票代號
            price: 新價格
        """
        old_price = self._prices.get(symbol, 0)
        self._prices[symbol] = price

        # 計算變動百分比
        if old_price > 0:
            change_percent = ((price - old_price) / old_price) * 100
        else:
            change_percent = 0

        # 建立並發送事件
        event_data = {
            'symbol': symbol,
            'old_price': old_price,
            'new_price': price,
            'change_percent': change_percent
        }

        # 根據變動幅度發送不同類型的事件
        if abs(change_percent) > 5:
            event = Event('price_alert', event_data)
        else:
            event = Event('price_update', event_data)

        self.notify(event)

    def get_price(self, symbol: str) -> float:
        """
        取得股票價格

        Args:
            symbol: 股票代號

        Returns:
            目前價格
        """
        return self._prices.get(symbol, 0)

class EmailNotifier(Observer):
    """
    電子郵件通知觀察者

    當收到事件時發送電子郵件通知
    """

    def __init__(self, email: str):
        """
        初始化

        Args:
            email: 電子郵件地址
        """
        self._email = email

    def update(self, event: Event):
        """
        處理事件

        Args:
            event: 事件物件
        """
        if event.event_type == 'price_alert':
            data = event.data
            message = (
                f"股價警示!{data['symbol']} 價格變動 "
                f"{data['change_percent']:.2f}%: "
                f"${data['old_price']:.2f} -> ${data['new_price']:.2f}"
            )
            print(f"[Email to {self._email}] {message}")
        else:
            print(f"[Email to {self._email}] 收到事件: {event.event_type}")

    def get_name(self) -> str:
        """
        傳回觀察者名稱
        """
        return f"EmailNotifier({self._email})"

class LoggingObserver(Observer):
    """
    日誌記錄觀察者

    將所有事件記錄到日誌
    """

    def __init__(self, log_name: str = "system"):
        """
        初始化

        Args:
            log_name: 日誌名稱
        """
        self._log_name = log_name
        self._logs: List[str] = []

    def update(self, event: Event):
        """
        記錄事件

        Args:
            event: 事件物件
        """
        log_entry = (
            f"[{event.timestamp.strftime('%Y-%m-%d %H:%M:%S')}] "
            f"{event.event_type}: {event.data}"
        )
        self._logs.append(log_entry)
        print(f"[Log:{self._log_name}] {log_entry}")

    def get_name(self) -> str:
        """
        傳回觀察者名稱
        """
        return f"LoggingObserver({self._log_name})"

    def get_logs(self) -> List[str]:
        """
        取得日誌記錄

        Returns:
            日誌列表
        """
        return self._logs.copy()

class DashboardUpdater(Observer):
    """
    儀表板更新觀察者

    更新即時儀表板顯示
    """

    def __init__(self, dashboard_name: str):
        """
        初始化

        Args:
            dashboard_name: 儀表板名稱
        """
        self._dashboard_name = dashboard_name
        self._current_data: Dict[str, Any] = {}

    def update(self, event: Event):
        """
        更新儀表板

        Args:
            event: 事件物件
        """
        if event.data and isinstance(event.data, dict):
            symbol = event.data.get('symbol', 'unknown')
            self._current_data[symbol] = event.data

            print(f"[Dashboard:{self._dashboard_name}] 更新 {symbol}: "
                  f"${event.data.get('new_price', 0):.2f}")

    def get_name(self) -> str:
        """
        傳回觀察者名稱
        """
        return f"DashboardUpdater({self._dashboard_name})"

    def get_current_data(self) -> Dict[str, Any]:
        """
        取得儀表板目前資料

        Returns:
            目前資料字典
        """
        return self._current_data.copy()

# 使用範例
if __name__ == "__main__":
    print("=== Observer 模式示範 ===\n")

    # 建立股票市場(主題)
    market = StockMarket()

    # 建立觀察者
    email_notifier = EmailNotifier("[email protected]")
    logger = LoggingObserver("trading")
    dashboard = DashboardUpdater("main")

    # 訂閱事件
    market.attach(logger)  # 訂閱所有事件
    market.attach(email_notifier, ['price_alert'])  # 只訂閱警示
    market.attach(dashboard, ['price_update', 'price_alert'])  # 訂閱價格相關

    print("\n--- 股票價格更新 ---\n")

    # 模擬價格變動
    market.set_price("AAPL", 150.00)  # 初始價格
    print()

    market.set_price("AAPL", 152.00)  # 小幅上漲
    print()

    market.set_price("AAPL", 165.00)  # 大幅上漲,觸發警示
    print()

    market.set_price("GOOGL", 2800.00)  # 另一支股票
    print()

    # 顯示儀表板資料
    print("\n--- 儀表板資料 ---")
    for symbol, data in dashboard.get_current_data().items():
        print(f"{symbol}: ${data['new_price']:.2f}")

設計模式的組合應用

在實際專案中,單一設計模式往往不足以解決複雜的問題。成熟的軟體系統通常會結合多種模式來建構更加靈活和強健的架構。以下範例展示如何組合多種模式來建構一個事件處理系統:

from abc import ABC, abstractmethod
from typing import Dict, List, Any, Callable
from datetime import datetime
import threading
import queue
import json

class EventHandler(ABC):
    """
    事件處理器抽象類別(Strategy 模式的策略介面)

    定義處理事件的介面
    """

    @abstractmethod
    def handle(self, event: Dict[str, Any]) -> bool:
        """
        處理事件

        Args:
            event: 事件資料

        Returns:
            是否處理成功
        """
        pass

    @abstractmethod
    def can_handle(self, event_type: str) -> bool:
        """
        檢查是否能處理特定類型的事件

        Args:
            event_type: 事件類型

        Returns:
            是否能處理
        """
        pass

class LogEventHandler(EventHandler):
    """
    日誌事件處理器

    將事件記錄到日誌
    """

    def __init__(self, log_file: str = "events.log"):
        """
        初始化日誌處理器

        Args:
            log_file: 日誌檔案名稱
        """
        self._log_file = log_file

    def handle(self, event: Dict[str, Any]) -> bool:
        """
        記錄事件到日誌

        Args:
            event: 事件資料

        Returns:
            是否成功
        """
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'event': event
        }
        print(f"[LOG] {json.dumps(log_entry, default=str)}")
        return True

    def can_handle(self, event_type: str) -> bool:
        """
        日誌處理器可以處理所有類型的事件

        Args:
            event_type: 事件類型

        Returns:
            永遠傳回 True
        """
        return True

class AlertEventHandler(EventHandler):
    """
    警示事件處理器

    處理需要發送警示的事件
    """

    def __init__(self, alert_types: List[str]):
        """
        初始化警示處理器

        Args:
            alert_types: 需要警示的事件類型列表
        """
        self._alert_types = alert_types

    def handle(self, event: Dict[str, Any]) -> bool:
        """
        發送警示

        Args:
            event: 事件資料

        Returns:
            是否成功
        """
        print(f"[ALERT] 警示事件: {event}")
        return True

    def can_handle(self, event_type: str) -> bool:
        """
        檢查是否為警示類型的事件

        Args:
            event_type: 事件類型

        Returns:
            是否為警示類型
        """
        return event_type in self._alert_types

class MetricsEventHandler(EventHandler):
    """
    指標事件處理器

    收集事件指標
    """

    def __init__(self):
        """
        初始化指標處理器
        """
        self._metrics: Dict[str, int] = {}

    def handle(self, event: Dict[str, Any]) -> bool:
        """
        更新指標

        Args:
            event: 事件資料

        Returns:
            是否成功
        """
        event_type = event.get('type', 'unknown')
        self._metrics[event_type] = self._metrics.get(event_type, 0) + 1
        return True

    def can_handle(self, event_type: str) -> bool:
        """
        指標處理器處理所有事件

        Args:
            event_type: 事件類型

        Returns:
            永遠傳回 True
        """
        return True

    def get_metrics(self) -> Dict[str, int]:
        """
        取得指標

        Returns:
            指標字典
        """
        return self._metrics.copy()

class EventProcessor:
    """
    事件處理器(組合 Strategy 和 Observer 模式)

    管理多個處理器並分發事件
    使用 Chain of Responsibility 模式處理事件
    """

    def __init__(self):
        """
        初始化事件處理器
        """
        self._handlers: List[EventHandler] = []
        self._event_queue = queue.Queue()
        self._running = False
        self._lock = threading.Lock()

    def register_handler(self, handler: EventHandler):
        """
        註冊事件處理器

        Args:
            handler: 要註冊的處理器
        """
        with self._lock:
            self._handlers.append(handler)
        print(f"已註冊處理器: {handler.__class__.__name__}")

    def process_event(self, event: Dict[str, Any]):
        """
        處理單一事件

        將事件分發給所有能處理它的處理器

        Args:
            event: 事件資料
        """
        event_type = event.get('type', 'unknown')

        with self._lock:
            handlers_copy = self._handlers.copy()

        for handler in handlers_copy:
            if handler.can_handle(event_type):
                try:
                    handler.handle(event)
                except Exception as e:
                    print(f"處理器 {handler.__class__.__name__} 發生錯誤: {e}")

    def submit_event(self, event: Dict[str, Any]):
        """
        提交事件到佇列

        Args:
            event: 事件資料
        """
        self._event_queue.put(event)

    def start_async_processing(self):
        """
        啟動非同步處理

        在背景執行緒中處理事件佇列
        """
        self._running = True

        def worker():
            while self._running:
                try:
                    event = self._event_queue.get(timeout=1)
                    self.process_event(event)
                    self._event_queue.task_done()
                except queue.Empty:
                    continue

        thread = threading.Thread(target=worker, daemon=True)
        thread.start()
        print("非同步事件處理已啟動")

    def stop(self):
        """
        停止處理
        """
        self._running = False
        print("事件處理已停止")

class EventProcessorBuilder:
    """
    事件處理器建構器(Builder 模式)

    提供流暢的 API 來建構事件處理器
    """

    def __init__(self):
        """
        初始化建構器
        """
        self._processor = EventProcessor()

    def with_logging(self, log_file: str = "events.log") -> 'EventProcessorBuilder':
        """
        新增日誌處理

        Args:
            log_file: 日誌檔案

        Returns:
            建構器本身
        """
        self._processor.register_handler(LogEventHandler(log_file))
        return self

    def with_alerts(self, alert_types: List[str]) -> 'EventProcessorBuilder':
        """
        新增警示處理

        Args:
            alert_types: 警示事件類型

        Returns:
            建構器本身
        """
        self._processor.register_handler(AlertEventHandler(alert_types))
        return self

    def with_metrics(self) -> 'EventProcessorBuilder':
        """
        新增指標收集

        Returns:
            建構器本身
        """
        handler = MetricsEventHandler()
        self._processor.register_handler(handler)
        self._metrics_handler = handler
        return self

    def with_custom_handler(self, handler: EventHandler) -> 'EventProcessorBuilder':
        """
        新增自訂處理器

        Args:
            handler: 自訂處理器

        Returns:
            建構器本身
        """
        self._processor.register_handler(handler)
        return self

    def build(self) -> EventProcessor:
        """
        建構並傳回處理器

        Returns:
            建構好的事件處理器
        """
        return self._processor

    def get_metrics_handler(self) -> MetricsEventHandler:
        """
        取得指標處理器

        Returns:
            指標處理器實例
        """
        return getattr(self, '_metrics_handler', None)

# 使用範例
if __name__ == "__main__":
    print("=== 設計模式組合應用示範 ===\n")

    # 使用 Builder 建構事件處理器
    builder = (EventProcessorBuilder()
               .with_logging()
               .with_alerts(['error', 'critical'])
               .with_metrics())

    processor = builder.build()
    metrics_handler = builder.get_metrics_handler()

    print("\n--- 處理事件 ---\n")

    # 處理各種事件
    events = [
        {'type': 'info', 'message': '系統啟動'},
        {'type': 'user_login', 'user': 'alice'},
        {'type': 'error', 'message': '連線失敗'},
        {'type': 'info', 'message': '處理完成'},
        {'type': 'critical', 'message': '資料庫無回應'},
    ]

    for event in events:
        processor.process_event(event)
        print()

    # 顯示指標
    if metrics_handler:
        print("--- 事件指標 ---")
        for event_type, count in metrics_handler.get_metrics().items():
            print(f"  {event_type}: {count}")

設計模式與 Python 特性的結合

Python 的動態特性為設計模式的實作提供了獨特的可能性。裝飾器、描述器、元類別等 Python 特性可以讓模式的實作更加優雅和 Pythonic。

使用描述器實作代理模式

Python 的描述器協議提供了一種優雅的方式來實作代理模式,特別是在控制屬性存取時:

from typing import Any, Optional
import time

class LazyProperty:
    """
    延遲載入屬性描述器

    實作代理模式,延遲計算昂貴的屬性值
    只在第一次存取時計算,之後使用快取值
    """

    def __init__(self, func):
        """
        初始化描述器

        Args:
            func: 計算屬性值的函式
        """
        self._func = func
        self._name = func.__name__

    def __get__(self, obj, objtype=None):
        """
        當屬性被存取時呼叫

        第一次存取時計算值並快取
        之後直接傳回快取值

        Args:
            obj: 實例物件
            objtype: 類別

        Returns:
            屬性值
        """
        if obj is None:
            # 透過類別存取,傳回描述器本身
            return self

        # 計算值
        value = self._func(obj)
        # 將值儲存在實例的 __dict__ 中
        # 這樣下次存取會直接從 __dict__ 取得,不會再呼叫描述器
        setattr(obj, self._name, value)
        return value

class ValidatedProperty:
    """
    驗證屬性描述器

    在設定值時進行驗證
    """

    def __init__(self, validator=None, default=None):
        """
        初始化驗證描述器

        Args:
            validator: 驗證函式,接受值並傳回 True/False
            default: 預設值
        """
        self._validator = validator
        self._default = default
        self._name = None

    def __set_name__(self, owner, name):
        """
        當描述器被指派給類別屬性時呼叫

        Args:
            owner: 擁有者類別
            name: 屬性名稱
        """
        self._name = f'_validated_{name}'

    def __get__(self, obj, objtype=None):
        """
        取得屬性值

        Args:
            obj: 實例
            objtype: 類別

        Returns:
            屬性值
        """
        if obj is None:
            return self
        return getattr(obj, self._name, self._default)

    def __set__(self, obj, value):
        """
        設定屬性值(帶驗證)

        Args:
            obj: 實例
            value: 要設定的值

        Raises:
            ValueError: 當驗證失敗時
        """
        if self._validator is not None:
            if not self._validator(value):
                raise ValueError(f"值 '{value}' 未通過驗證")
        setattr(obj, self._name, value)

class DataModel:
    """
    資料模型類別

    展示描述器的使用
    """

    # 使用驗證描述器
    name = ValidatedProperty(
        validator=lambda x: isinstance(x, str) and len(x) > 0,
        default=""
    )

    age = ValidatedProperty(
        validator=lambda x: isinstance(x, int) and 0 <= x <= 150,
        default=0
    )

    email = ValidatedProperty(
        validator=lambda x: isinstance(x, str) and '@' in x,
        default=""
    )

    def __init__(self, name: str, age: int, email: str):
        """
        初始化資料模型

        Args:
            name: 姓名
            age: 年齡
            email: 電子郵件
        """
        self.name = name
        self.age = age
        self.email = email

    @LazyProperty
    def computed_hash(self):
        """
        計算雜湊值(延遲計算的昂貴操作)

        Returns:
            計算的雜湊值
        """
        print("計算雜湊值(這是一個昂貴的操作)...")
        time.sleep(0.5)  # 模擬昂貴的計算
        return hash((self.name, self.age, self.email))

    def __repr__(self):
        """
        字串表示
        """
        return f"DataModel(name={self.name}, age={self.age}, email={self.email})"

# 使用範例
if __name__ == "__main__":
    print("=== Python 描述器與設計模式 ===\n")

    # 建立資料模型
    model = DataModel("Alice", 30, "[email protected]")
    print(f"建立模型: {model}")

    # 延遲載入屬性
    print("\n第一次存取 computed_hash:")
    hash1 = model.computed_hash
    print(f"雜湊值: {hash1}")

    print("\n第二次存取 computed_hash(使用快取):")
    hash2 = model.computed_hash
    print(f"雜湊值: {hash2}")

    # 驗證屬性
    print("\n--- 屬性驗證 ---")
    try:
        model.age = 200  # 應該失敗
    except ValueError as e:
        print(f"驗證失敗: {e}")

    try:
        model.email = "invalid"  # 應該失敗
    except ValueError as e:
        print(f"驗證失敗: {e}")

    # 成功的更新
    model.age = 31
    print(f"年齡更新成功: {model.age}")

設計模式的最佳實踐

在實際開發中運用設計模式時,需要注意幾個重要的原則。首先是不要為了使用模式而使用模式。設計模式是解決特定問題的工具,如果問題不存在,使用模式只會增加不必要的複雜度。其次是要理解模式背後的原則,而不只是機械地套用模式的結構。當你理解了開閉原則、依賴反轉原則等核心概念後,就能更靈活地應用和調整模式。

另一個重要的考量是模式的組合。複雜的系統往往需要多個模式協同工作,理解模式之間的關係和互補性是進階開發者的必備能力。例如,Factory 模式經常與 Singleton 模式結合使用,Strategy 模式常與 Template Method 模式搭配,Observer 模式則常與 Mediator 模式一起使用來管理複雜的物件互動。

最後,要記得 Python 的特性可能讓某些傳統模式變得不必要或可以大幅簡化。例如,Python 的一級函式讓 Command 模式可以用簡單的函式物件實作,而不需要建立額外的類別。Python 的鴨子型別也讓許多基於介面的模式實作變得更加簡潔。善用這些特性,可以讓你的程式碼既符合設計模式的精神,又保持 Python 的簡潔優雅。

設計模式的學習是一個持續的過程。隨著經驗的累積,你會越來越能夠識別何時該使用哪個模式,以及如何根據具體情況調整模式的實作。最重要的是保持務實的態度,始終以解決實際問題為目標,而不是追求模式的完美應用。