第3章抽象工厂模式

学习目标

完成本章学习后，读者将能够：

深入理解抽象工厂模式的理论基础、产品族概念与形式化定义
掌握Python语言特性下的多种抽象工厂实现机制
分析抽象工厂与工厂方法模式的本质区别与适用场景
设计跨平台、跨主题、跨数据库的产品族创建体系
评估抽象工厂模式在软件架构中的利弊与替代方案

3.1 理论基础与模式定义

3.1.1 形式化定义

抽象工厂模式（Abstract Factory Pattern） 提供一个创建一系列相关或相互依赖对象的接口，而无需指定它们具体的类。其核心约束可形式化表述为：

$$\text{AbstractFactory} = {f_1, f_2, ..., f_n}$$

其中每个工厂方法 $f_i$ 创建特定类型的产品：

$$f_i: \text{Factory} \rightarrow \text{Product}_i$$

产品族约束：对于任意具体工厂 $F \in \text{ConcreteFactory}$，其创建的所有产品必须属于同一产品族：

$$\forall F, \forall p_1, p_2 \in \text{Products}(F): \text{Family}(p_1) = \text{Family}(p_2)$$

产品变体约束：不同工厂创建的同类产品属于不同变体：

$$\forall F_1 \neq F_2, \forall p_1 \in \text{Products}(F_1), p_2 \in \text{Products}(F_2): \text{Type}(p_1) = \text{Type}(p_2) \Rightarrow \text{Variant}(p_1) \neq \text{Variant}(p_2)$$

3.1.2 历史演进与学术背景

抽象工厂模式的概念可追溯至建筑师Christopher Alexander的模式语言理论，后被GoF系统化为创建型设计模式。该模式解决了产品族创建这一软件工程中的核心问题：

┌─────────────────────────────────────────────────────────────────┐
│                    产品族与产品变体                              │
│                                                                 │
│                    产品类型 (Product Types)                      │
│              ┌──────────┬──────────┬──────────┐                 │
│              │ Button   │ Checkbox │ Input    │                 │
│  产品族      ├──────────┼──────────┼──────────┤                 │
│  (Family)    │          │          │          │                 │
│  ──────────  │ Windows  │ WinBtn   │ WinChk   │ WinInp │        │
│              │──────────┼──────────┼──────────┼────────│        │
│              │ macOS    │ MacBtn   │ MacChk   │ MacInp │        │
│              │──────────┼──────────┼──────────┼────────│        │
│              │ Linux    │ LnxBtn   │ LnxChk   │ LnxInp │        │
│              └──────────┴──────────┴──────────┴────────┘        │
│                                                                 │
│  每一行是一个产品族（由一个工厂创建）                            │
│  每一列是一个产品类型（由不同的工厂方法创建）                    │
└─────────────────────────────────────────────────────────────────┘

3.1.3 模式动机与应用场景

抽象工厂模式解决的核心问题是产品族的一致性约束。考虑以下典型场景：

应用场景	产品族	产品类型	一致性约束
跨平台UI	Windows/macOS/Linux	Button, Checkbox, Input	视觉风格一致
数据库访问	MySQL/PostgreSQL/Oracle	Connection, Command, Transaction	协议兼容
主题系统	Light/Dark/Custom	Button, Input, Card	颜色协调
支付网关	Stripe/PayPal/Alipay	Processor, Subscription, Webhook	API兼容
游戏引擎	OpenGL/DirectX/Vulkan	Renderer, Texture, Shader	渲染管线兼容

核心设计动机：

产品一致性：确保同一族的产品能够协同工作
隔离具体类：客户端与具体产品类解耦
易于切换：更换产品族只需更换工厂
开闭原则：新增产品族无需修改客户端代码

3.1.4 UML结构模型

┌─────────────────────────────────────────────────────────────────┐
│                  AbstractFactory «interface»                     │
├─────────────────────────────────────────────────────────────────┤
│  + create_product_a(): AbstractProductA                         │
│  + create_product_b(): AbstractProductB                         │
│  + create_product_c(): AbstractProductC                         │
└─────────────────────────────────────────────────────────────────┘
                           △
                           │ implements
              ┌────────────┴────────────┐
              │                         │
┌─────────────┴─────────┐   ┌──────────┴──────────┐
│   ConcreteFactory1    │   │   ConcreteFactory2  │
├───────────────────────┤   ├─────────────────────┤
│  + create_product_a() │   │  + create_product_a │
│  + create_product_b() │   │  + create_product_b │
│  + create_product_c() │   │  + create_product_c │
└───────────────────────┘   └─────────────────────┘
         │                           │
         │ creates                   │ creates
         ▼                           ▼
┌─────────────────┐         ┌─────────────────┐
│ ProductA1       │         │ ProductA2       │
│ ProductB1       │         │ ProductB2       │
│ ProductC1       │         │ ProductC2       │
│ (Family 1)      │         │ (Family 2)      │
└─────────────────┘         └─────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                  AbstractProductA «interface»                    │
├─────────────────────────────────────────────────────────────────┤
│  + operation_a(): Result                                        │
└─────────────────────────────────────────────────────────────────┘
                           △
              ┌────────────┴────────────┐
              │                         │
┌─────────────┴─────────┐   ┌──────────┴──────────┐
│   ConcreteProductA1   │   │   ConcreteProductA2 │
├───────────────────────┤   ├─────────────────────┤
│  + operation_a()      │   │  + operation_a()    │
└───────────────────────┘   └─────────────────────┘

结构要素解析：

角色	职责	数量关系
AbstractFactory	声明创建抽象产品的方法	1个
ConcreteFactory	实现创建具体产品的方法	每个产品族1个
AbstractProduct	声明产品接口	每个产品类型1个
ConcreteProduct	实现产品接口	产品族数 × 产品类型数
Client	使用抽象工厂和抽象产品	1个或多个

3.2 Python实现机制深度解析

3.2.1 基于抽象基类的经典实现

python

from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Dict, Type, Any, Optional
from dataclasses import dataclass, field
from enum import Enum, auto

class ProductA(ABC):
    """产品A抽象基类"""
    
    @abstractmethod
    def operation_a(self) -> str:
        """产品A的操作"""
        pass
    
    @abstractmethod
    def get_info(self) -> Dict[str, Any]:
        """获取产品信息"""
        pass


class ProductB(ABC):
    """产品B抽象基类"""
    
    @abstractmethod
    def operation_b(self) -> str:
        """产品B的操作"""
        pass
    
    @abstractmethod
    def collaborate(self, collaborator: ProductA) -> str:
        """与产品A协作"""
        pass


class ConcreteProductA1(ProductA):
    """具体产品A1（属于产品族1）"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self._config = config or {}
        self._family = "Family1"
    
    def operation_a(self) -> str:
        return f"[{self._family}] ProductA1 执行操作"
    
    def get_info(self) -> Dict[str, Any]:
        return {
            "type": "ProductA1",
            "family": self._family,
            "config": self._config
        }


class ConcreteProductA2(ProductA):
    """具体产品A2（属于产品族2）"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self._config = config or {}
        self._family = "Family2"
    
    def operation_a(self) -> str:
        return f"[{self._family}] ProductA2 执行操作"
    
    def get_info(self) -> Dict[str, Any]:
        return {
            "type": "ProductA2",
            "family": self._family,
            "config": self._config
        }


class ConcreteProductB1(ProductB):
    """具体产品B1（属于产品族1）"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self._config = config or {}
        self._family = "Family1"
    
    def operation_b(self) -> str:
        return f"[{self._family}] ProductB1 执行操作"
    
    def collaborate(self, collaborator: ProductA) -> str:
        return f"[{self._family}] ProductB1 与 {collaborator.operation_a()} 协作"


class ConcreteProductB2(ProductB):
    """具体产品B2（属于产品族2）"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self._config = config or {}
        self._family = "Family2"
    
    def operation_b(self) -> str:
        return f"[{self._family}] ProductB2 执行操作"
    
    def collaborate(self, collaborator: ProductA) -> str:
        return f"[{self._family}] ProductB2 与 {collaborator.operation_a()} 协作"


class AbstractFactory(ABC):
    """抽象工厂基类"""
    
    @abstractmethod
    def create_product_a(self, **kwargs) -> ProductA:
        """创建产品A"""
        pass
    
    @abstractmethod
    def create_product_b(self, **kwargs) -> ProductB:
        """创建产品B"""
        pass
    
    @property
    @abstractmethod
    def family_name(self) -> str:
        """产品族名称"""
        pass


class ConcreteFactory1(AbstractFactory):
    """具体工厂1：创建产品族1的所有产品"""
    
    @property
    def family_name(self) -> str:
        return "Family1"
    
    def create_product_a(self, **kwargs) -> ProductA:
        return ConcreteProductA1(kwargs)
    
    def create_product_b(self, **kwargs) -> ProductB:
        return ConcreteProductB1(kwargs)


class ConcreteFactory2(AbstractFactory):
    """具体工厂2：创建产品族2的所有产品"""
    
    @property
    def family_name(self) -> str:
        return "Family2"
    
    def create_product_a(self, **kwargs) -> ProductA:
        return ConcreteProductA2(kwargs)
    
    def create_product_b(self, **kwargs) -> ProductB:
        return ConcreteProductB2(kwargs)


def client_code(factory: AbstractFactory) -> None:
    """
    客户端代码
    
    关键点：
        1. 客户端只依赖抽象工厂和抽象产品
        2. 产品族一致性由工厂保证
        3. 切换产品族只需更换工厂实例
    """
    print(f"使用工厂: {factory.family_name}")
    
    product_a = factory.create_product_a(setting="value")
    product_b = factory.create_product_b(option="enabled")
    
    print(product_a.operation_a())
    print(product_b.operation_b())
    print(product_b.collaborate(product_a))
    print()


client_code(ConcreteFactory1())
client_code(ConcreteFactory2())

3.2.2 基于Protocol的现代实现

python

from typing import Protocol, runtime_checkable, ClassVar, Dict, Any, Type
from dataclasses import dataclass, field

@runtime_checkable
class Button(Protocol):
    """按钮协议"""
    
    label: str
    
    def render(self) -> str:
        """渲染按钮"""
        ...
    
    def on_click(self) -> str:
        """点击事件"""
        ...


@runtime_checkable
class Checkbox(Protocol):
    """复选框协议"""
    
    checked: bool
    
    def render(self) -> str:
        """渲染复选框"""
        ...
    
    def toggle(self) -> None:
        """切换状态"""
        ...


@runtime_checkable
class Input(Protocol):
    """输入框协议"""
    
    value: str
    placeholder: str
    
    def render(self) -> str:
        """渲染输入框"""
        ...
    
    def get_value(self) -> str:
        """获取值"""
        ...


@dataclass
class WindowsButton:
    """Windows风格按钮"""
    
    label: str = "Button"
    _style: str = field(default="windows", repr=False)
    
    def render(self) -> str:
        return f"[Windows Button] {self.label}"
    
    def on_click(self) -> str:
        return f"[Windows] Button '{self.label}' clicked"


@dataclass
class WindowsCheckbox:
    """Windows风格复选框"""
    
    checked: bool = False
    _style: str = field(default="windows", repr=False)
    
    def render(self) -> str:
        state = "☑" if self.checked else "☐"
        return f"[Windows Checkbox] {state}"
    
    def toggle(self) -> None:
        self.checked = not self.checked


@dataclass
class WindowsInput:
    """Windows风格输入框"""
    
    value: str = ""
    placeholder: str = "Enter text..."
    _style: str = field(default="windows", repr=False)
    
    def render(self) -> str:
        return f"[Windows Input] {self.value or self.placeholder}"
    
    def get_value(self) -> str:
        return self.value


@dataclass
class MacOSButton:
    """macOS风格按钮"""
    
    label: str = "Button"
    _style: str = field(default="macos", repr=False)
    
    def render(self) -> str:
        return f"[macOS Button] {self.label}"
    
    def on_click(self) -> str:
        return f"[macOS] Button '{self.label}' clicked"


@dataclass
class MacOSCheckbox:
    """macOS风格复选框"""
    
    checked: bool = False
    _style: str = field(default="macos", repr=False)
    
    def render(self) -> str:
        state = "✓" if self.checked else "○"
        return f"[macOS Checkbox] {state}"
    
    def toggle(self) -> None:
        self.checked = not self.checked


@dataclass
class MacOSInput:
    """macOS风格输入框"""
    
    value: str = ""
    placeholder: str = "Enter text..."
    _style: str = field(default="macos", repr=False)
    
    def render(self) -> str:
        return f"[macOS Input] {self.value or self.placeholder}"
    
    def get_value(self) -> str:
        return self.value


class GUIFactory(Protocol):
    """GUI工厂协议"""
    
    def create_button(self, label: str = "Button") -> Button:
        """创建按钮"""
        ...
    
    def create_checkbox(self, checked: bool = False) -> Checkbox:
        """创建复选框"""
        ...
    
    def create_input(self, placeholder: str = "") -> Input:
        """创建输入框"""
        ...
    
    @property
    def platform(self) -> str:
        """平台名称"""
        ...


@dataclass
class WindowsFactory:
    """Windows GUI工厂"""
    
    @property
    def platform(self) -> str:
        return "Windows"
    
    def create_button(self, label: str = "Button") -> Button:
        return WindowsButton(label=label)
    
    def create_checkbox(self, checked: bool = False) -> Checkbox:
        return WindowsCheckbox(checked=checked)
    
    def create_input(self, placeholder: str = "") -> Input:
        return WindowsInput(placeholder=placeholder)


@dataclass
class MacOSFactory:
    """macOS GUI工厂"""
    
    @property
    def platform(self) -> str:
        return "macOS"
    
    def create_button(self, label: str = "Button") -> Button:
        return MacOSButton(label=label)
    
    def create_checkbox(self, checked: bool = False) -> Checkbox:
        return MacOSCheckbox(checked=checked)
    
    def create_input(self, placeholder: str = "") -> Input:
        return MacOSInput(placeholder=placeholder)


class Application:
    """
    应用程序
    
    特性:
        - 使用抽象工厂创建UI组件
        - 所有组件属于同一产品族（保证一致性）
        - 切换平台只需更换工厂
    """
    
    def __init__(self, factory: GUIFactory):
        self._factory = factory
        self._button = factory.create_button("Submit")
        self._checkbox = factory.create_checkbox()
        self._input = factory.create_input("Enter your name")
    
    def render(self) -> None:
        """渲染所有组件"""
        print(f"Platform: {self._factory.platform}")
        print(self._button.render())
        print(self._checkbox.render())
        print(self._input.render())
    
    def interact(self) -> None:
        """模拟交互"""
        print(self._button.on_click())
        self._checkbox.toggle()
        print(self._checkbox.render())


def get_factory(platform: str) -> GUIFactory:
    """根据平台获取工厂"""
    factories: Dict[str, Type[GUIFactory]] = {
        "windows": WindowsFactory,
        "macos": MacOSFactory,
    }
    
    if platform not in factories:
        raise ValueError(f"不支持的平台: {platform}")
    
    return factories[platform]()


import platform
system = platform.system().lower()
detected = "windows" if system == "windows" else "macos" if system == "darwin" else "windows"

factory = get_factory(detected)
app = Application(factory)
app.render()
print()
app.interact()

3.2.3 配置驱动的工厂选择

python

from typing import Dict, Type, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum, auto
import json
import threading

class FactoryType(Enum):
    """工厂类型枚举"""
    WINDOWS = auto()
    MACOS = auto()
    LINUX = auto()
    MYSQL = auto()
    POSTGRESQL = auto()
    STRIPE = auto()
    PAYPAL = auto()


@dataclass
class FactoryConfig:
    """工厂配置"""
    factory_type: FactoryType
    settings: Dict[str, Any] = field(default_factory=dict)
    enabled: bool = True


class FactoryRegistry:
    """
    工厂注册表
    
    特性:
        - 集中管理所有工厂
        - 配置驱动工厂选择
        - 支持运行时切换
    """
    
    _factories: Dict[FactoryType, Type] = {}
    _configs: Dict[FactoryType, FactoryConfig] = {}
    _instances: Dict[FactoryType, Any] = {}
    _lock = threading.RLock()
    
    @classmethod
    def register(cls, factory_type: FactoryType, 
                factory_class: Type) -> None:
        """注册工厂类"""
        with cls._lock:
            cls._factories[factory_type] = factory_class
    
    @classmethod
    def configure(cls, config: FactoryConfig) -> None:
        """配置工厂"""
        with cls._lock:
            cls._configs[config.factory_type] = config
    
    @classmethod
    def get_factory(cls, factory_type: FactoryType) -> Any:
        """获取工厂实例（单例）"""
        with cls._lock:
            if factory_type not in cls._factories:
                raise ValueError(f"未注册的工厂类型: {factory_type}")
            
            if factory_type not in cls._instances:
                factory_class = cls._factories[factory_type]
                config = cls._configs.get(factory_type)
                kwargs = config.settings if config else {}
                cls._instances[factory_type] = factory_class(**kwargs)
            
            return cls._instances[factory_type]
    
    @classmethod
    def from_config_file(cls, file_path: str) -> None:
        """从配置文件加载"""
        with open(file_path, 'r') as f:
            config_data = json.load(f)
        
        for name, settings in config_data.get("factories", {}).items():
            try:
                factory_type = FactoryType[name.upper()]
                cls.configure(FactoryConfig(
                    factory_type=factory_type,
                    settings=settings
                ))
            except KeyError:
                print(f"警告: 未知的工厂类型 {name}")
    
    @classmethod
    def list_available(cls) -> list:
        """列出可用的工厂"""
        with cls._lock:
            return [ft.name for ft in cls._factories.keys()]


class ConfigurableFactory:
    """可配置的抽象工厂基类"""
    
    def __init__(self, **kwargs):
        self._config = kwargs
    
    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> 'ConfigurableFactory':
        """从字典创建"""
        return cls(**config)

3.3 实际应用案例

3.3.1 企业级数据库访问工厂

python

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List, ContextManager
from dataclasses import dataclass, field
from contextlib import contextmanager
from enum import Enum, auto
import threading
import time

class DatabaseType(Enum):
    MYSQL = auto()
    POSTGRESQL = auto()
    SQLITE = auto()
    MONGODB = auto()


@dataclass
class ConnectionConfig:
    """连接配置"""
    host: str = "localhost"
    port: int = 3306
    database: str = "test"
    username: str = "root"
    password: str = ""
    charset: str = "utf8mb4"
    pool_size: int = 5
    timeout: float = 30.0


@dataclass
class QueryResult:
    """查询结果"""
    success: bool
    data: Any = None
    affected_rows: int = 0
    error: Optional[str] = None
    execution_time: float = 0.0


class Connection(ABC):
    """连接抽象"""
    
    def __init__(self, config: ConnectionConfig):
        self._config = config
        self._connected = False
    
    @abstractmethod
    def connect(self) -> bool:
        pass
    
    @abstractmethod
    def disconnect(self) -> None:
        pass
    
    @abstractmethod
    def is_connected(self) -> bool:
        pass
    
    @property
    def config(self) -> ConnectionConfig:
        return self._config


class Command(ABC):
    """命令抽象"""
    
    @abstractmethod
    def execute(self, sql: str, params: tuple = None) -> QueryResult:
        pass
    
    @abstractmethod
    def execute_scalar(self, sql: str, params: tuple = None) -> Any:
        pass
    
    @abstractmethod
    def execute_many(self, sql: str, params_list: List[tuple]) -> int:
        pass


class Transaction(ABC):
    """事务抽象"""
    
    @abstractmethod
    def begin(self) -> None:
        pass
    
    @abstractmethod
    def commit(self) -> None:
        pass
    
    @abstractmethod
    def rollback(self) -> None:
        pass
    
    @contextmanager
    def scope(self):
        """事务作用域"""
        self.begin()
        try:
            yield self
            self.commit()
        except Exception:
            self.rollback()
            raise


class DataAdapter(ABC):
    """数据适配器抽象"""
    
    @abstractmethod
    def to_dict(self, result: QueryResult) -> List[Dict]:
        pass
    
    @abstractmethod
    def from_dict(self, data: Dict) -> str:
        pass


class MySQLConnection(Connection):
    """MySQL连接"""
    
    def connect(self) -> bool:
        print(f"[MySQL] 连接到 {self._config.host}:{self._config.port}/{self._config.database}")
        self._connected = True
        return True
    
    def disconnect(self) -> None:
        print("[MySQL] 断开连接")
        self._connected = False
    
    def is_connected(self) -> bool:
        return self._connected


class MySQLCommand(Command):
    """MySQL命令"""
    
    def __init__(self, connection: MySQLConnection):
        self._connection = connection
    
    def execute(self, sql: str, params: tuple = None) -> QueryResult:
        start = time.time()
        print(f"[MySQL] 执行: {sql}")
        return QueryResult(
            success=True,
            data=[{"id": 1, "name": "test"}],
            affected_rows=1,
            execution_time=time.time() - start
        )
    
    def execute_scalar(self, sql: str, params: tuple = None) -> Any:
        result = self.execute(sql, params)
        if result.data:
            return list(result.data[0].values())[0]
        return None
    
    def execute_many(self, sql: str, params_list: List[tuple]) -> int:
        print(f"[MySQL] 批量执行: {sql}, {len(params_list)} 条")
        return len(params_list)


class MySQLTransaction(Transaction):
    """MySQL事务"""
    
    def begin(self) -> None:
        print("[MySQL] 事务开始")
    
    def commit(self) -> None:
        print("[MySQL] 事务提交")
    
    def rollback(self) -> None:
        print("[MySQL] 事务回滚")


class MySQLDataAdapter(DataAdapter):
    """MySQL数据适配器"""
    
    def to_dict(self, result: QueryResult) -> List[Dict]:
        return result.data if result.data else []
    
    def from_dict(self, data: Dict) -> str:
        columns = ", ".join(data.keys())
        values = ", ".join(f"'{v}'" for v in data.values())
        return f"INSERT INTO table ({columns}) VALUES ({values})"


class PostgreSQLConnection(Connection):
    """PostgreSQL连接"""
    
    def connect(self) -> bool:
        print(f"[PostgreSQL] 连接到 {self._config.host}:{self._config.port}/{self._config.database}")
        self._connected = True
        return True
    
    def disconnect(self) -> None:
        print("[PostgreSQL] 断开连接")
        self._connected = False
    
    def is_connected(self) -> bool:
        return self._connected


class PostgreSQLCommand(Command):
    """PostgreSQL命令"""
    
    def __init__(self, connection: PostgreSQLConnection):
        self._connection = connection
    
    def execute(self, sql: str, params: tuple = None) -> QueryResult:
        start = time.time()
        print(f"[PostgreSQL] 执行: {sql}")
        return QueryResult(
            success=True,
            data=[{"id": 1, "name": "test"}],
            affected_rows=1,
            execution_time=time.time() - start
        )
    
    def execute_scalar(self, sql: str, params: tuple = None) -> Any:
        result = self.execute(sql, params)
        if result.data:
            return list(result.data[0].values())[0]
        return None
    
    def execute_many(self, sql: str, params_list: List[tuple]) -> int:
        print(f"[PostgreSQL] 批量执行: {sql}, {len(params_list)} 条")
        return len(params_list)


class PostgreSQLTransaction(Transaction):
    """PostgreSQL事务"""
    
    def begin(self) -> None:
        print("[PostgreSQL] 事务开始")
    
    def commit(self) -> None:
        print("[PostgreSQL] 事务提交")
    
    def rollback(self) -> None:
        print("[PostgreSQL] 事务回滚")


class PostgreSQLDataAdapter(DataAdapter):
    """PostgreSQL数据适配器"""
    
    def to_dict(self, result: QueryResult) -> List[Dict]:
        return result.data if result.data else []
    
    def from_dict(self, data: Dict) -> str:
        columns = ", ".join(data.keys())
        placeholders = ", ".join(f"%({k})s" for k in data.keys())
        return f"INSERT INTO table ({columns}) VALUES ({placeholders})"


class DatabaseFactory(ABC):
    """
    数据库抽象工厂
    
    产品族:
        - Connection: 数据库连接
        - Command: SQL命令执行器
        - Transaction: 事务管理器
        - DataAdapter: 数据适配器
    """
    
    @abstractmethod
    def create_connection(self, config: ConnectionConfig) -> Connection:
        pass
    
    @abstractmethod
    def create_command(self, connection: Connection) -> Command:
        pass
    
    @abstractmethod
    def create_transaction(self) -> Transaction:
        pass
    
    @abstractmethod
    def create_adapter(self) -> DataAdapter:
        pass
    
    @property
    @abstractmethod
    def database_type(self) -> DatabaseType:
        pass


class MySQLFactory(DatabaseFactory):
    """MySQL工厂"""
    
    @property
    def database_type(self) -> DatabaseType:
        return DatabaseType.MYSQL
    
    def create_connection(self, config: ConnectionConfig) -> Connection:
        return MySQLConnection(config)
    
    def create_command(self, connection: Connection) -> Command:
        return MySQLCommand(connection)
    
    def create_transaction(self) -> Transaction:
        return MySQLTransaction()
    
    def create_adapter(self) -> DataAdapter:
        return MySQLDataAdapter()


class PostgreSQLFactory(DatabaseFactory):
    """PostgreSQL工厂"""
    
    @property
    def database_type(self) -> DatabaseType:
        return DatabaseType.POSTGRESQL
    
    def create_connection(self, config: ConnectionConfig) -> Connection:
        return PostgreSQLConnection(config)
    
    def create_command(self, connection: Connection) -> Command:
        return PostgreSQLCommand(connection)
    
    def create_transaction(self) -> Transaction:
        return PostgreSQLTransaction()
    
    def create_adapter(self) -> DataAdapter:
        return PostgreSQLDataAdapter()


class DatabaseClient:
    """
    数据库客户端
    
    使用抽象工厂创建的所有产品
    """
    
    def __init__(self, factory: DatabaseFactory, config: ConnectionConfig):
        self._factory = factory
        self._config = config
        self._connection: Optional[Connection] = None
        self._command: Optional[Command] = None
        self._adapter = factory.create_adapter()
    
    def connect(self) -> bool:
        self._connection = self._factory.create_connection(self._config)
        self._command = self._factory.create_command(self._connection)
        return self._connection.connect()
    
    def disconnect(self) -> None:
        if self._connection:
            self._connection.disconnect()
    
    def execute(self, sql: str, params: tuple = None) -> QueryResult:
        if not self._command:
            raise RuntimeError("未连接数据库")
        return self._command.execute(sql, params)
    
    @contextmanager
    def transaction(self):
        """事务上下文"""
        tx = self._factory.create_transaction()
        with tx.scope():
            yield tx
    
    def insert(self, table: str, data: Dict) -> QueryResult:
        sql = self._adapter.from_dict(data).replace("table", table)
        return self.execute(sql)


config = ConnectionConfig(
    host="localhost",
    port=3306,
    database="production",
    username="admin"
)

client = DatabaseClient(MySQLFactory(), config)
client.connect()

with client.transaction():
    result = client.execute("SELECT * FROM users")
    print(f"查询结果: {result.data}")

client.disconnect()

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
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381

3.3.2 多主题UI组件工厂

python

from typing import Protocol, Dict, Any, Tuple, runtime_checkable
from dataclasses import dataclass, field
from enum import Enum, auto

class ThemeType(Enum):
    LIGHT = auto()
    DARK = auto()
    HIGH_CONTRAST = auto()
    OCEAN = auto()


@dataclass
class Color:
    """颜色值"""
    r: int
    g: int
    b: int
    a: float = 1.0
    
    def to_hex(self) -> str:
        return f"#{self.r:02x}{self.g:02x}{self.b:02x}"
    
    def to_rgba(self) -> str:
        return f"rgba({self.r}, {self.g}, {self.b}, {self.a})"
    
    def lighten(self, percent: float) -> 'Color':
        factor = percent / 100
        return Color(
            r=min(255, int(self.r + (255 - self.r) * factor)),
            g=min(255, int(self.g + (255 - self.g) * factor)),
            b=min(255, int(self.b + (255 - self.b) * factor)),
            a=self.a
        )
    
    def darken(self, percent: float) -> 'Color':
        factor = 1 - percent / 100
        return Color(
            r=int(self.r * factor),
            g=int(self.g * factor),
            b=int(self.b * factor),
            a=self.a
        )


@dataclass
class Typography:
    """排版样式"""
    font_family: str
    font_size: int
    font_weight: int
    line_height: float
    
    def to_css(self) -> str:
        return f"font: {self.font_weight} {self.font_size}px/{self.line_height} {self.font_family}"


@dataclass
class Spacing:
    """间距"""
    padding: Tuple[int, int, int, int]
    margin: Tuple[int, int, int, int]
    border_radius: int


class ThemeElement(Protocol):
    """主题元素协议"""
    
    @property
    def background_color(self) -> Color:
        ...
    
    @property
    def text_color(self) -> Color:
        ...
    
    @property
    def border_color(self) -> Color:
        ...
    
    def get_styles(self) -> Dict[str, Any]:
        ...


@dataclass
class ButtonTheme:
    """按钮主题"""
    
    _background: Color
    _text: Color
    _border: Color
    _hover_bg: Color
    _active_bg: Color
    
    @property
    def background_color(self) -> Color:
        return self._background
    
    @property
    def text_color(self) -> Color:
        return self._text
    
    @property
    def border_color(self) -> Color:
        return self._border
    
    @property
    def hover_background(self) -> Color:
        return self._hover_bg
    
    @property
    def active_background(self) -> Color:
        return self._active_bg
    
    def get_styles(self) -> Dict[str, Any]:
        return {
            "background": self._background.to_hex(),
            "color": self._text.to_hex(),
            "border": f"1px solid {self._border.to_hex()}",
            "hover": {"background": self._hover_bg.to_hex()},
            "active": {"background": self._active_bg.to_hex()}
        }


@dataclass
class InputTheme:
    """输入框主题"""
    
    _background: Color
    _text: Color
    _border: Color
    _placeholder: Color
    _focus_border: Color
    
    @property
    def background_color(self) -> Color:
        return self._background
    
    @property
    def text_color(self) -> Color:
        return self._text
    
    @property
    def border_color(self) -> Color:
        return self._border
    
    @property
    def placeholder_color(self) -> Color:
        return self._placeholder
    
    @property
    def focus_border_color(self) -> Color:
        return self._focus_border
    
    def get_styles(self) -> Dict[str, Any]:
        return {
            "background": self._background.to_hex(),
            "color": self._text.to_hex(),
            "border": f"1px solid {self._border.to_hex()}",
            "placeholder": {"color": self._placeholder.to_hex()},
            "focus": {"borderColor": self._focus_border.to_hex()}
        }


@dataclass
class CardTheme:
    """卡片主题"""
    
    _background: Color
    _text: Color
    _border: Color
    _shadow: str
    
    @property
    def background_color(self) -> Color:
        return self._background
    
    @property
    def text_color(self) -> Color:
        return self._text
    
    @property
    def border_color(self) -> Color:
        return self._border
    
    @property
    def shadow(self) -> str:
        return self._shadow
    
    def get_styles(self) -> Dict[str, Any]:
        return {
            "background": self._background.to_hex(),
            "color": self._text.to_hex(),
            "border": f"1px solid {self._border.to_hex()}",
            "boxShadow": self._shadow
        }


class ThemeFactory(ABC):
    """
    主题抽象工厂
    
    产品族:
        - ButtonTheme: 按钮样式
        - InputTheme: 输入框样式
        - CardTheme: 卡片样式
    """
    
    @abstractmethod
    def create_button_theme(self) -> ButtonTheme:
        pass
    
    @abstractmethod
    def create_input_theme(self) -> InputTheme:
        pass
    
    @abstractmethod
    def create_card_theme(self) -> CardTheme:
        pass
    
    @property
    @abstractmethod
    def theme_name(self) -> str:
        pass
    
    @property
    @abstractmethod
    def theme_type(self) -> ThemeType:
        pass
    
    @property
    @abstractmethod
    def primary_color(self) -> Color:
        pass
    
    @property
    @abstractmethod
    def background_color(self) -> Color:
        pass
    
    @property
    @abstractmethod
    def text_color(self) -> Color:
        pass


class LightThemeFactory(ThemeFactory):
    """浅色主题工厂"""
    
    @property
    def theme_name(self) -> str:
        return "Light Theme"
    
    @property
    def theme_type(self) -> ThemeType:
        return ThemeType.LIGHT
    
    @property
    def primary_color(self) -> Color:
        return Color(59, 130, 246)
    
    @property
    def background_color(self) -> Color:
        return Color(255, 255, 255)
    
    @property
    def text_color(self) -> Color:
        return Color(17, 24, 39)
    
    def create_button_theme(self) -> ButtonTheme:
        return ButtonTheme(
            _background=self.primary_color,
            _text=Color(255, 255, 255),
            _border=self.primary_color,
            _hover_bg=self.primary_color.darken(10),
            _active_bg=self.primary_color.darken(20)
        )
    
    def create_input_theme(self) -> InputTheme:
        return InputTheme(
            _background=Color(255, 255, 255),
            _text=self.text_color,
            _border=Color(209, 213, 219),
            _placeholder=Color(156, 163, 175),
            _focus_border=self.primary_color
        )
    
    def create_card_theme(self) -> CardTheme:
        return CardTheme(
            _background=Color(255, 255, 255),
            _text=self.text_color,
            _border=Color(229, 231, 235),
            _shadow="0 1px 3px rgba(0, 0, 0, 0.1)"
        )


class DarkThemeFactory(ThemeFactory):
    """深色主题工厂"""
    
    @property
    def theme_name(self) -> str:
        return "Dark Theme"
    
    @property
    def theme_type(self) -> ThemeType:
        return ThemeType.DARK
    
    @property
    def primary_color(self) -> Color:
        return Color(96, 165, 250)
    
    @property
    def background_color(self) -> Color:
        return Color(17, 24, 39)
    
    @property
    def text_color(self) -> Color:
        return Color(243, 244, 246)
    
    def create_button_theme(self) -> ButtonTheme:
        return ButtonTheme(
            _background=self.primary_color,
            _text=Color(17, 24, 39),
            _border=self.primary_color,
            _hover_bg=self.primary_color.lighten(10),
            _active_bg=self.primary_color.lighten(20)
        )
    
    def create_input_theme(self) -> InputTheme:
        return InputTheme(
            _background=Color(31, 41, 55),
            _text=self.text_color,
            _border=Color(55, 65, 81),
            _placeholder=Color(107, 114, 128),
            _focus_border=self.primary_color
        )
    
    def create_card_theme(self) -> CardTheme:
        return CardTheme(
            _background=Color(31, 41, 55),
            _text=self.text_color,
            _border=Color(55, 65, 81),
            _shadow="0 1px 3px rgba(0, 0, 0, 0.3)"
        )


class ThemeManager:
    """主题管理器"""
    
    _current_factory: Optional[ThemeFactory] = None
    _listeners: list = []
    
    @classmethod
    def set_theme(cls, factory: ThemeFactory) -> None:
        cls._current_factory = factory
        cls._notify_listeners()
    
    @classmethod
    def get_factory(cls) -> Optional[ThemeFactory]:
        return cls._current_factory
    
    @classmethod
    def add_listener(cls, callback: Callable) -> None:
        cls._listeners.append(callback)
    
    @classmethod
    def _notify_listeners(cls) -> None:
        for listener in cls._listeners:
            listener(cls._current_factory)


ThemeManager.set_theme(DarkThemeFactory())

factory = ThemeManager.get_factory()
print(f"当前主题: {factory.theme_name}")
print(f"按钮样式: {factory.create_button_theme().get_styles()}")
print(f"输入框样式: {factory.create_input_theme().get_styles()}")

3.4 抽象工厂 vs 工厂方法

3.4.1 核心区别

┌─────────────────────────────────────────────────────────────────┐
│                    工厂方法 vs 抽象工厂                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  工厂方法                                                       │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  目的: 创建单一产品                                      │   │
│  │  结构: Creator + Product                                 │   │
│  │  扩展: 新增产品类型需要新增Creator                        │   │
│  │  关系: 一对一 (一个工厂创建一个产品)                      │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│  抽象工厂                                                       │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  目的: 创建产品族                                        │   │
│  │  结构: AbstractFactory + 多个Product                     │   │
│  │  扩展: 新增产品族需要新增Factory                          │   │
│  │  关系: 一对多 (一个工厂创建多个相关产品)                  │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

3.4.2 模式协作关系

抽象工厂通常使用工厂方法来实现：

python

class AbstractFactory(ABC):
    @abstractmethod
    def create_product_a(self) -> ProductA:
        pass
    
    @abstractmethod
    def create_product_b(self) -> ProductB:
        pass

class ConcreteFactory1(AbstractFactory):
    def create_product_a(self) -> ProductA:
        return ConcreteProductA1()
    
    def create_product_b(self) -> ProductB:
        return ConcreteProductB1()

每个create_product_x方法本质上是一个工厂方法。

3.4.3 选择决策表

场景特征	推荐模式
只需创建一种产品	工厂方法
需要创建多个相关产品	抽象工厂
产品之间有协作关系	抽象工厂
需要保证产品族一致性	抽象工厂
产品类型经常变化	工厂方法
产品族经常变化	抽象工厂

3.5 反模式与最佳实践

3.5.1 常见反模式

反模式1：产品族划分不当

python

class BadFactory:
    """错误示例：将不相关的产品放入同一产品族"""
    
    def create_button(self) -> Button:
        return WindowsButton()
    
    def create_database_connection(self) -> Connection:
        return MySQLConnection()

问题：按钮和数据库连接没有一致性约束，不应属于同一产品族。

反模式2：过度抽象

python

class OverAbstractFactory:
    """错误示例：为每个产品创建单独的工厂"""
    
    def get_button_factory(self) -> ButtonFactory:
        return WindowsButtonFactory()
    
    def get_input_factory(self) -> InputFactory:
        return WindowsInputFactory()

问题：失去了抽象工厂的产品族一致性保证。

3.5.2 最佳实践

python

from typing import Protocol, TypeVar, Generic, Dict, Type

T = TypeVar('T')

class ProductFamily(Protocol):
    """产品族协议"""
    
    @property
    def family_name(self) -> str:
        ...


class AbstractFactoryBase(ABC, Generic[T]):
    """
    泛型抽象工厂基类
    
    特性:
        - 类型安全的产品创建
        - 产品族一致性保证
        - 可扩展的产品类型
    """
    
    _product_registry: Dict[str, Type] = {}
    
    @classmethod
    def register_product(cls, product_type: str, product_class: Type) -> None:
        cls._product_registry[product_type] = product_class
    
    @property
    @abstractmethod
    def family_name(self) -> str:
        pass
    
    def create(self, product_type: str, **kwargs) -> T:
        if product_type not in self._product_registry:
            raise ValueError(f"未知产品类型: {product_type}")
        
        product_class = self._product_registry[product_type]
        return product_class(family=self.family_name, **kwargs)

3.6 模式评估与决策指南

3.6.1 适用性检查清单

检查项	是	否	说明
需要创建多个相关产品？	✓ 使用	✗ 评估	产品族特征
产品之间有一致性约束？	✓ 使用	✗ 评估	产品协作
需要支持多种产品族？	✓ 使用	✗ 评估	可切换性
产品族相对稳定？	✓ 使用	✗ 评估	扩展成本
产品类型可能变化？	✗ 评估	✓ 使用	考虑工厂方法

3.6.2 扩展成本分析

扩展类型	抽象工厂	工厂方法
新增产品族	新增1个工厂类	新增N个工厂类
新增产品类型	修改所有工厂	新增1个工厂类

3.7 小结

抽象工厂模式通过产品族抽象解决了相关对象的一致性创建问题：

┌─────────────────────────────────────────────────────────────────┐
│                    抽象工厂模式决策树                            │
│                                                                 │
│                    需要创建多个产品？                            │
│                         │                                       │
│              ┌──────────┴──────────┐                            │
│              ▼                     ▼                            │
│            是的                   不需要                         │
│              │                     │                            │
│              ▼                     ▼                            │
│        产品有协作关系？        工厂方法                          │
│              │                                                  │
│      ┌───────┴───────┐                                          │
│      ▼               ▼                                          │
│     是              否                                           │
│      │               │                                          │
│      ▼               ▼                                          │
│   抽象工厂       独立工厂方法                                     │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

关键要点：

核心价值：保证产品族一致性，隔离具体产品类
产品族概念：一组协同工作的相关产品
扩展权衡：新增产品族容易，新增产品类型困难
Python特性：利用Protocol、dataclass简化实现
模式协作：抽象工厂通常使用工厂方法实现

在下一章中，我们将探讨建造者模式，学习如何分步骤构建复杂对象。

第3章 抽象工厂模式 ​

学习目标 ​

3.1 理论基础与模式定义 ​

3.1.1 形式化定义 ​

3.1.2 历史演进与学术背景 ​

3.1.3 模式动机与应用场景 ​

3.1.4 UML结构模型 ​

3.2 Python实现机制深度解析 ​

3.2.1 基于抽象基类的经典实现 ​

3.2.2 基于Protocol的现代实现 ​

3.2.3 配置驱动的工厂选择 ​

3.3 实际应用案例 ​

3.3.1 企业级数据库访问工厂 ​

3.3.2 多主题UI组件工厂 ​

3.4 抽象工厂 vs 工厂方法 ​

3.4.1 核心区别 ​

3.4.2 模式协作关系 ​

3.4.3 选择决策表 ​

3.5 反模式与最佳实践 ​

3.5.1 常见反模式 ​

3.5.2 最佳实践 ​

3.6 模式评估与决策指南 ​

3.6.1 适用性检查清单 ​

3.6.2 扩展成本分析 ​

3.7 小结 ​

第3章抽象工厂模式

学习目标

3.1 理论基础与模式定义

3.1.1 形式化定义

3.1.2 历史演进与学术背景

3.1.3 模式动机与应用场景

3.1.4 UML结构模型

3.2 Python实现机制深度解析

3.2.1 基于抽象基类的经典实现

3.2.2 基于Protocol的现代实现

3.2.3 配置驱动的工厂选择

3.3 实际应用案例

3.3.1 企业级数据库访问工厂

3.3.2 多主题UI组件工厂

3.4 抽象工厂 vs 工厂方法

3.4.1 核心区别

3.4.2 模式协作关系

3.4.3 选择决策表

3.5 反模式与最佳实践

3.5.1 常见反模式

3.5.2 最佳实践

3.6 模式评估与决策指南

3.6.1 适用性检查清单

3.6.2 扩展成本分析

3.7 小结