第4章 建造者模式
学习目标
完成本章学习后,读者将能够:
- 深入理解建造者模式的理论基础、设计动机与形式化定义
- 掌握Python语言特性下的多种建造者实现机制及其设计权衡
- 分析建造者模式与工厂模式、原型模式的本质区别与适用场景
- 设计复杂对象的分步构建体系与链式API
- 评估建造者模式在软件架构中的利弊与替代方案
4.1 理论基础与模式定义
4.1.1 形式化定义
建造者模式(Builder Pattern) 将一个复杂对象的构建与它的表示分离,使得同样的构建过程可以创建不同的表示。其核心思想可形式化表述为:
$$\text{Product} = \text{Builder}.\text{build_part}_1() \circ \text{build_part}_2() \circ ... \circ \text{build_part}_n() \circ \text{get_result}()$$
其中 $\circ$ 表示构建步骤的组合操作。该模式满足以下约束:
构建独立性约束:
$$\forall b \in \text{Builder}, \forall p_1, p_2 \in \text{Parts}: \text{build}(p_1) \perp \text{build}(p_2)$$
即各部件的构建相互独立,构建顺序不影响最终结果(在语义等价的前提下)。
表示多样性约束:
$$\forall b_1, b_2 \in \text{ConcreteBuilder}, b_1 \neq b_2 \Rightarrow \text{Result}(b_1) \neq \text{Result}(b_2)$$
即不同的具体建造者产生不同的产品表示。
4.1.2 历史演进与学术背景
建造者模式源于对复杂对象创建问题的抽象。在软件工程实践中,存在一类对象具有以下特征:
- 多部件组成:由多个子部件构成
- 构建复杂:需要多个步骤完成构建
- 表示多样:同一构建过程可产生不同表示
- 验证需求:构建过程需要验证约束条件
┌─────────────────────────────────────────────────────────────────┐
│ 构造函数 vs 建造者模式 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 构造函数方式 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Computer(cpu, ram, storage, gpu, motherboard, │ │
│ │ power_supply, case, cooling, peripherals) │ │
│ │ │ │
│ │ 问题: │ │
│ │ - 参数列表过长,难以理解 │ │
│ │ - 可选参数需要传递None │ │
│ │ - 参数顺序容易混淆 │ │
│ │ - 无法进行中间验证 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ 建造者模式 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ComputerBuilder() │ │
│ │ .set_cpu("i9") │ │
│ │ .set_ram("64GB") │ │
│ │ .set_storage("2TB") │ │
│ │ .build() │ │
│ │ │ │
│ │ 优势: │ │
│ │ - 链式调用,语义清晰 │ │
│ │ - 可选参数自然省略 │ │
│ │ - 命名明确,不易混淆 │ │
│ │ - 可在build()时验证 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘4.1.3 模式动机与应用场景
建造者模式解决的核心问题是复杂对象的分步构建。考虑以下典型场景:
| 应用场景 | 复杂性来源 | 建造者优势 |
|---|---|---|
| 计算机配置 | 多种硬件组合 | 链式配置、验证兼容性 |
| SQL查询 | 动态条件组合 | 安全构建、防止注入 |
| HTTP请求 | 多种请求选项 | 流畅API、类型安全 |
| 文档生成 | 结构化内容 | 分步构建、模板支持 |
| 游戏角色 | 多属性配置 | 默认值、验证规则 |
| 配置对象 | 嵌套配置 | 层级构建、环境适配 |
核心设计动机:
- 分离构建与表示:相同的构建过程创建不同的表示
- 分步构建:将复杂构建分解为简单步骤
- 参数命名:通过方法名明确参数含义
- 验证机制:在构建时验证约束条件
- 不可变对象:构建完成后对象不可修改
4.1.4 UML结构模型
┌─────────────────────────────────────────────────────────────────┐
│ Director «optional» │
├─────────────────────────────────────────────────────────────────┤
│ - builder: Builder │
│ + construct(): Product │
│ │
│ // 封装标准构建流程 │
│ construct(): │
│ builder.build_part_a() │
│ builder.build_part_b() │
│ builder.build_part_c() │
│ return builder.get_result() │
└─────────────────────────────────────────────────────────────────┘
│ uses
▼
┌─────────────────────────────────────────────────────────────────┐
│ Builder «interface» │
├─────────────────────────────────────────────────────────────────┤
│ + build_part_a(): void │
│ + build_part_b(): void │
│ + build_part_c(): void │
│ + get_result(): Product │
└─────────────────────────────────────────────────────────────────┘
△
│ implements
┌────────────┴────────────┐
│ │
┌─────────────┴─────────┐ ┌──────────┴──────────┐
│ ConcreteBuilder1 │ │ ConcreteBuilder2 │
├───────────────────────┤ ├─────────────────────┤
│ - product: Product1 │ │ - product: Product2│
│ + build_part_a() │ │ + build_part_a() │
│ + build_part_b() │ │ + build_part_b() │
│ + build_part_c() │ │ + build_part_c() │
│ + get_result() │ │ + get_result() │
└───────────────────────┘ └─────────────────────┘
│ │
│ creates │ creates
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ Product1 │ │ Product2 │
├─────────────────┤ ├─────────────────┤
│ - part_a │ │ - part_a │
│ - part_b │ │ - part_b │
│ - part_c │ │ - part_c │
└─────────────────┘ └─────────────────┘结构要素解析:
| 角色 | 职责 | 是否必需 |
|---|---|---|
| Builder | 声明构建步骤的抽象接口 | 必需 |
| ConcreteBuilder | 实现具体构建步骤 | 必需 |
| Product | 被构建的复杂对象 | 必需 |
| Director | 封装标准构建流程 | 可选 |
4.2 Python实现机制深度解析
4.2.1 经典建造者模式
from abc import ABC, abstractmethod
from typing import Optional, List, Dict, Any, TypeVar, Generic
from dataclasses import dataclass, field
from enum import Enum, auto
T = TypeVar('T')
@dataclass
class Computer:
"""计算机产品"""
cpu: str = ""
ram: str = ""
storage: str = ""
gpu: Optional[str] = None
motherboard: str = ""
power_supply: str = ""
case: str = ""
cooling: str = ""
peripherals: List[str] = field(default_factory=list)
def get_specs(self) -> Dict[str, Any]:
return {
"CPU": self.cpu,
"RAM": self.ram,
"Storage": self.storage,
"GPU": self.gpu or "集成显卡",
"Motherboard": self.motherboard,
"Power": self.power_supply,
"Case": self.case,
"Cooling": self.cooling,
"Peripherals": self.peripherals
}
class ComputerBuilder(ABC):
"""计算机建造者抽象基类"""
@abstractmethod
def build_cpu(self, cpu: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_ram(self, ram: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_storage(self, storage: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_gpu(self, gpu: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_motherboard(self, motherboard: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_power_supply(self, power_supply: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_case(self, case: str) -> 'ComputerBuilder':
pass
@abstractmethod
def build_cooling(self, cooling: str) -> 'ComputerBuilder':
pass
@abstractmethod
def add_peripheral(self, peripheral: str) -> 'ComputerBuilder':
pass
@abstractmethod
def get_result(self) -> Computer:
pass
class GamingComputerBuilder(ComputerBuilder):
"""游戏电脑建造者"""
def __init__(self):
self._computer = Computer()
def build_cpu(self, cpu: str) -> 'GamingComputerBuilder':
self._computer.cpu = cpu
return self
def build_ram(self, ram: str) -> 'GamingComputerBuilder':
self._computer.ram = ram
return self
def build_storage(self, storage: str) -> 'GamingComputerBuilder':
self._computer.storage = storage
return self
def build_gpu(self, gpu: str) -> 'GamingComputerBuilder':
self._computer.gpu = gpu
return self
def build_motherboard(self, motherboard: str) -> 'GamingComputerBuilder':
self._computer.motherboard = motherboard
return self
def build_power_supply(self, power_supply: str) -> 'GamingComputerBuilder':
self._computer.power_supply = power_supply
return self
def build_case(self, case: str) -> 'GamingComputerBuilder':
self._computer.case = case
return self
def build_cooling(self, cooling: str) -> 'GamingComputerBuilder':
self._computer.cooling = cooling
return self
def add_peripheral(self, peripheral: str) -> 'GamingComputerBuilder':
self._computer.peripherals.append(peripheral)
return self
def get_result(self) -> Computer:
self._validate()
return self._computer
def _validate(self):
if not self._computer.cpu:
raise ValueError("CPU是必需的")
if not self._computer.ram:
raise ValueError("RAM是必需的")
if not self._computer.gpu:
raise ValueError("游戏电脑必须配置独立显卡")
class OfficeComputerBuilder(ComputerBuilder):
"""办公电脑建造者"""
def __init__(self):
self._computer = Computer()
def build_cpu(self, cpu: str) -> 'OfficeComputerBuilder':
self._computer.cpu = cpu
return self
def build_ram(self, ram: str) -> 'OfficeComputerBuilder':
self._computer.ram = ram
return self
def build_storage(self, storage: str) -> 'OfficeComputerBuilder':
self._computer.storage = storage
return self
def build_gpu(self, gpu: str) -> 'OfficeComputerBuilder':
self._computer.gpu = gpu
return self
def build_motherboard(self, motherboard: str) -> 'OfficeComputerBuilder':
self._computer.motherboard = motherboard
return self
def build_power_supply(self, power_supply: str) -> 'OfficeComputerBuilder':
self._computer.power_supply = power_supply
return self
def build_case(self, case: str) -> 'OfficeComputerBuilder':
self._computer.case = case
return self
def build_cooling(self, cooling: str) -> 'OfficeComputerBuilder':
self._computer.cooling = cooling
return self
def add_peripheral(self, peripheral: str) -> 'OfficeComputerBuilder':
self._computer.peripherals.append(peripheral)
return self
def get_result(self) -> Computer:
self._validate()
return self._computer
def _validate(self):
if not self._computer.cpu:
raise ValueError("CPU是必需的")
if not self._computer.ram:
raise ValueError("RAM是必需的")
class ComputerDirector:
"""
计算机指挥者
职责:
- 封装标准构建流程
- 提供预设配置方案
"""
def __init__(self, builder: ComputerBuilder):
self._builder = builder
def build_high_end_gaming(self) -> Computer:
"""构建高端游戏电脑"""
return (self._builder
.build_cpu("Intel i9-13900K")
.build_ram("64GB DDR5-6000")
.build_storage("2TB Samsung 990 Pro NVMe")
.build_gpu("NVIDIA RTX 4090 24GB")
.build_motherboard("ASUS ROG Maximus Z790")
.build_power_supply("1200W 80+ Titanium")
.build_case("Lian Li O11 Dynamic EVO")
.build_cooling("360mm AIO + 6x 120mm Fans")
.add_peripheral("ROG Azoth Keyboard")
.add_peripheral("ROG Gladius III Mouse")
.add_peripheral("ROG Swift PG32UQ Monitor")
.get_result())
def build_mid_range_gaming(self) -> Computer:
"""构建中端游戏电脑"""
return (self._builder
.build_cpu("AMD Ryzen 7 7800X3D")
.build_ram("32GB DDR5-5200")
.build_storage("1TB WD Black SN850X")
.build_gpu("NVIDIA RTX 4070 Ti")
.build_motherboard("MSI MAG B650 TOMAHAWK")
.build_power_supply("750W 80+ Gold")
.build_case("Corsair 4000D Airflow")
.build_cooling("240mm AIO")
.get_result())
def build_office_workstation(self) -> Computer:
"""构建办公工作站"""
return (self._builder
.build_cpu("Intel i5-13600")
.build_ram("16GB DDR4-3200")
.build_storage("512GB NVMe SSD")
.build_motherboard("ASUS Prime B760M-A")
.build_power_supply("500W 80+ Bronze")
.build_case("Fractal Design Define Mini")
.build_cooling("Stock Cooler")
.add_peripheral("Logitech MX Keys")
.add_peripheral("Logitech MX Master 3")
.get_result())
gaming_builder = GamingComputerBuilder()
director = ComputerDirector(gaming_builder)
high_end = director.build_high_end_gaming()
print("高端游戏电脑配置:")
for key, value in high_end.get_specs().items():
print(f" {key}: {value}")4.2.2 流畅建造者模式(Fluent Builder)
流畅建造者模式利用Python的链式调用特性,提供更简洁的API:
from typing import Optional, List, Dict, Any, Callable, TypeVar
from dataclasses import dataclass, field
from functools import wraps
T = TypeVar('T')
def step(func: Callable) -> Callable:
"""步骤装饰器:自动返回self以支持链式调用"""
@wraps(func)
def wrapper(self, *args, **kwargs):
func(self, *args, **kwargs)
return self
return wrapper
@dataclass
class HttpRequest:
"""HTTP请求"""
method: str
url: str
headers: Dict[str, str] = field(default_factory=dict)
query_params: Dict[str, str] = field(default_factory=dict)
body: Optional[Any] = None
timeout: float = 30.0
retry_count: int = 0
retry_delay: float = 1.0
verify_ssl: bool = True
follow_redirects: bool = True
def to_curl(self) -> str:
"""生成curl命令"""
parts = [f"curl -X {self.method}"]
for key, value in self.headers.items():
parts.append(f"-H '{key}: {value}'")
if self.body:
import json
body_str = json.dumps(self.body) if isinstance(self.body, dict) else str(self.body)
parts.append(f"-d '{body_str}'")
url = self.url
if self.query_params:
query = "&".join(f"{k}={v}" for k, v in self.query_params.items())
url = f"{url}?{query}"
parts.append(f"'{url}'")
return " ".join(parts)
class RequestBuilder:
"""
HTTP请求建造者
特性:
- 流畅API设计
- 类型安全
- 验证机制
- curl命令生成
"""
def __init__(self, url: str):
self._url = url
self._method: str = "GET"
self._headers: Dict[str, str] = {}
self._query_params: Dict[str, str] = {}
self._body: Optional[Any] = None
self._timeout: float = 30.0
self._retry_count: int = 0
self._retry_delay: float = 1.0
self._verify_ssl: bool = True
self._follow_redirects: bool = True
self._validators: List[Callable] = []
def get(self) -> 'RequestBuilder':
self._method = "GET"
return self
def post(self) -> 'RequestBuilder':
self._method = "POST"
return self
def put(self) -> 'RequestBuilder':
self._method = "PUT"
return self
def delete(self) -> 'RequestBuilder':
self._method = "DELETE"
return self
def patch(self) -> 'RequestBuilder':
self._method = "PATCH"
return self
def method(self, method: str) -> 'RequestBuilder':
self._method = method.upper()
return self
def header(self, key: str, value: str) -> 'RequestBuilder':
self._headers[key] = value
return self
def headers(self, headers: Dict[str, str]) -> 'RequestBuilder':
self._headers.update(headers)
return self
def content_type(self, content_type: str) -> 'RequestBuilder':
return self.header("Content-Type", content_type)
def accept(self, accept: str) -> 'RequestBuilder':
return self.header("Accept", accept)
def json_content(self) -> 'RequestBuilder':
return self.content_type("application/json").accept("application/json")
def authorization(self, token: str, scheme: str = "Bearer") -> 'RequestBuilder':
return self.header("Authorization", f"{scheme} {token}")
def basic_auth(self, username: str, password: str) -> 'RequestBuilder':
import base64
credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
return self.header("Authorization", f"Basic {credentials}")
def api_key(self, key: str, header_name: str = "X-API-Key") -> 'RequestBuilder':
return self.header(header_name, key)
def query(self, key: str, value: str) -> 'RequestBuilder':
self._query_params[key] = value
return self
def queries(self, params: Dict[str, str]) -> 'RequestBuilder':
self._query_params.update(params)
return self
def body(self, body: Any) -> 'RequestBuilder':
self._body = body
return self
def json_body(self, data: Dict[str, Any]) -> 'RequestBuilder':
self._body = data
return self.json_content()
def form_body(self, data: Dict[str, str]) -> 'RequestBuilder':
self._body = data
return self.content_type("application/x-www-form-urlencoded")
def timeout(self, seconds: float) -> 'RequestBuilder':
self._timeout = seconds
return self
def retry(self, count: int, delay: float = 1.0) -> 'RequestBuilder':
self._retry_count = count
self._retry_delay = delay
return self
def verify_ssl(self, verify: bool) -> 'RequestBuilder':
self._verify_ssl = verify
return self
def follow_redirects(self, follow: bool) -> 'RequestBuilder':
self._follow_redirects = follow
return self
def validate(self, validator: Callable[['HttpRequest'], bool]) -> 'RequestBuilder':
self._validators.append(validator)
return self
def build(self) -> HttpRequest:
request = HttpRequest(
method=self._method,
url=self._url,
headers=self._headers.copy(),
query_params=self._query_params.copy(),
body=self._body,
timeout=self._timeout,
retry_count=self._retry_count,
retry_delay=self._retry_delay,
verify_ssl=self._verify_ssl,
follow_redirects=self._follow_redirects
)
for validator in self._validators:
if not validator(request):
raise ValueError("请求验证失败")
return request
request = (RequestBuilder("https://api.example.com/v1/users")
.post()
.json_body({
"name": "张三",
"email": "zhangsan@example.com",
"role": "admin"
})
.authorization("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...")
.query("notify", "true")
.timeout(60.0)
.retry(3, delay=2.0)
.build())
print(f"方法: {request.method}")
print(f"URL: {request.url}")
print(f"curl: {request.to_curl()}")4.2.3 类型安全的泛型建造者
from typing import Generic, TypeVar, Callable, Optional, Dict, Any, List
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
T = TypeVar('T')
B = TypeVar('B', bound='BaseBuilder')
@dataclass
class BuildContext:
"""构建上下文"""
started_at: float = field(default_factory=lambda: __import__('time').time())
steps_completed: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
def record_step(self, step_name: str):
self.steps_completed.append(step_name)
def add_warning(self, message: str):
self.warnings.append(message)
class BaseBuilder(ABC, Generic[T]):
"""
建造者基类
提供通用的构建基础设施:
- 构建上下文管理
- 验证器注册
- 后处理器支持
"""
def __init__(self):
self._context = BuildContext()
self._validators: List[Callable[[T], bool]] = []
self._post_processors: List[Callable[[T], T]] = []
self._required_fields: List[str] = []
def require(self, field_name: str) -> 'BaseBuilder':
"""标记必需字段"""
self._required_fields.append(field_name)
return self
def validate(self, validator: Callable[[T], bool]) -> 'BaseBuilder':
"""添加验证器"""
self._validators.append(validator)
return self
def post_process(self, processor: Callable[[T], T]) -> 'BaseBuilder':
"""添加后处理器"""
self._post_processors.append(processor)
return self
def _record_step(self, step_name: str):
"""记录构建步骤"""
self._context.record_step(step_name)
def _add_warning(self, message: str):
"""添加警告"""
self._context.add_warning(message)
@abstractmethod
def _create_product(self) -> T:
"""创建产品实例"""
pass
@abstractmethod
def _get_field_value(self, field_name: str) -> Any:
"""获取字段值"""
pass
def build(self) -> T:
"""构建产品"""
for field_name in self._required_fields:
if not self._get_field_value(field_name):
raise ValueError(f"必需字段 '{field_name}' 未设置")
product = self._create_product()
for validator in self._validators:
if not validator(product):
raise ValueError("产品验证失败")
for processor in self._post_processors:
product = processor(product)
return product
def get_context(self) -> BuildContext:
"""获取构建上下文"""
return self._context
@dataclass
class DatabaseConfig:
"""数据库配置"""
host: str = "localhost"
port: int = 3306
database: str = ""
username: str = ""
password: str = ""
pool_size: int = 5
max_overflow: int = 10
pool_timeout: float = 30.0
charset: str = "utf8mb4"
ssl_enabled: bool = False
ssl_ca: Optional[str] = None
def to_connection_string(self) -> str:
return f"mysql://{self.username}:***@{self.host}:{self.port}/{self.database}"
class DatabaseConfigBuilder(BaseBuilder[DatabaseConfig]):
"""数据库配置建造者"""
def __init__(self):
super().__init__()
self._host: str = "localhost"
self._port: int = 3306
self._database: str = ""
self._username: str = ""
self._password: str = ""
self._pool_size: int = 5
self._max_overflow: int = 10
self._pool_timeout: float = 30.0
self._charset: str = "utf8mb4"
self._ssl_enabled: bool = False
self._ssl_ca: Optional[str] = None
self.require("database").require("username").require("password")
def host(self, host: str) -> 'DatabaseConfigBuilder':
self._host = host
self._record_step("host")
return self
def port(self, port: int) -> 'DatabaseConfigBuilder':
self._port = port
self._record_step("port")
return self
def database(self, database: str) -> 'DatabaseConfigBuilder':
self._database = database
self._record_step("database")
return self
def credentials(self, username: str, password: str) -> 'DatabaseConfigBuilder':
self._username = username
self._password = password
self._record_step("credentials")
return self
def pool(self, size: int, max_overflow: int = 10, timeout: float = 30.0) -> 'DatabaseConfigBuilder':
self._pool_size = size
self._max_overflow = max_overflow
self._pool_timeout = timeout
self._record_step("pool")
return self
def charset(self, charset: str) -> 'DatabaseConfigBuilder':
self._charset = charset
return self
def ssl(self, ca_path: str = None) -> 'DatabaseConfigBuilder':
self._ssl_enabled = True
self._ssl_ca = ca_path
self._record_step("ssl")
return self
def _create_product(self) -> DatabaseConfig:
return DatabaseConfig(
host=self._host,
port=self._port,
database=self._database,
username=self._username,
password=self._password,
pool_size=self._pool_size,
max_overflow=self._max_overflow,
pool_timeout=self._pool_timeout,
charset=self._charset,
ssl_enabled=self._ssl_enabled,
ssl_ca=self._ssl_ca
)
def _get_field_value(self, field_name: str) -> Any:
return getattr(self, f'_{field_name}', None)
config = (DatabaseConfigBuilder()
.host("192.168.1.100")
.port(3306)
.database("production_db")
.credentials("admin", "secret_password")
.pool(size=20, max_overflow=10, timeout=60.0)
.ssl("/path/to/ca.pem")
.validate(lambda c: c.port > 0 and c.port < 65536)
.build())
print(f"连接字符串: {config.to_connection_string()}")
print(f"连接池大小: {config.pool_size}")
print(f"SSL启用: {config.ssl_enabled}")4.2.4 不可变对象建造者
from typing import Optional, List, Dict, Any, Tuple
from dataclasses import dataclass, field
from functools import cached_property
import hashlib
import json
@dataclass(frozen=True)
class ImmutableDocument:
"""
不可变文档对象
特性:
- 创建后不可修改
- 支持哈希和比较
- 缓存计算属性
"""
title: str
content: str
author: str
tags: Tuple[str, ...] = field(default_factory=tuple)
metadata: Tuple[Tuple[str, str], ...] = field(default_factory=tuple)
version: int = 1
created_at: str = field(default_factory=lambda: __import__('datetime').datetime.now().isoformat())
@cached_property
def content_hash(self) -> str:
"""内容哈希值"""
return hashlib.sha256(self.content.encode()).hexdigest()[:16]
@cached_property
def word_count(self) -> int:
"""字数统计"""
return len(self.content.split())
@cached_property
def tag_set(self) -> set:
"""标签集合"""
return set(self.tags)
def to_dict(self) -> Dict[str, Any]:
return {
"title": self.title,
"content": self.content,
"author": self.author,
"tags": list(self.tags),
"metadata": dict(self.metadata),
"version": self.version,
"created_at": self.created_at,
"content_hash": self.content_hash
}
def __hash__(self):
return hash((self.title, self.content, self.author, self.tags, self.version))
class DocumentBuilder:
"""
不可变文档建造者
特性:
- 构建不可变对象
- 支持从现有文档创建变体
- 自动计算派生属性
"""
def __init__(self):
self._title: str = ""
self._content: str = ""
self._author: str = ""
self._tags: List[str] = []
self._metadata: Dict[str, str] = {}
self._version: int = 1
self._created_at: Optional[str] = None
@classmethod
def from_document(cls, document: ImmutableDocument) -> 'DocumentBuilder':
"""从现有文档创建建造者"""
builder = cls()
builder._title = document.title
builder._content = document.content
builder._author = document.author
builder._tags = list(document.tags)
builder._metadata = dict(document.metadata)
builder._version = document.version
builder._created_at = document.created_at
return builder
def title(self, title: str) -> 'DocumentBuilder':
self._title = title
return self
def content(self, content: str) -> 'DocumentBuilder':
self._content = content
return self
def author(self, author: str) -> 'DocumentBuilder':
self._author = author
return self
def add_tag(self, tag: str) -> 'DocumentBuilder':
if tag not in self._tags:
self._tags.append(tag)
return self
def tags(self, *tags: str) -> 'DocumentBuilder':
for tag in tags:
self.add_tag(tag)
return self
def metadata(self, key: str, value: str) -> 'DocumentBuilder':
self._metadata[key] = value
return self
def version(self, version: int) -> 'DocumentBuilder':
self._version = version
return self
def increment_version(self) -> 'DocumentBuilder':
self._version += 1
return self
def build(self) -> ImmutableDocument:
if not self._title:
raise ValueError("标题不能为空")
if not self._content:
raise ValueError("内容不能为空")
if not self._author:
raise ValueError("作者不能为空")
return ImmutableDocument(
title=self._title,
content=self._content,
author=self._author,
tags=tuple(self._tags),
metadata=tuple(self._metadata.items()),
version=self._version,
created_at=self._created_at
)
doc = (DocumentBuilder()
.title("Python设计模式详解")
.content("设计模式是软件工程中的重要概念..." * 100)
.author("张三")
.tags("Python", "设计模式", "软件工程", "编程")
.metadata("category", "技术文档")
.metadata("difficulty", "中级")
.build())
print(f"标题: {doc.title}")
print(f"字数: {doc.word_count}")
print(f"内容哈希: {doc.content_hash}")
print(f"标签: {doc.tag_set}")
doc_v2 = (DocumentBuilder.from_document(doc)
.content(doc.content + "\n\n新增内容...")
.increment_version()
.build())
print(f"\n版本: {doc_v2.version}")
print(f"哈希变化: {doc.content_hash} -> {doc_v2.content_hash}")4.3 实际应用案例
4.3.1 企业级SQL查询构建器
from typing import Optional, List, Tuple, Any, Union
from dataclasses import dataclass, field
from enum import Enum, auto
class JoinType(Enum):
INNER = "INNER JOIN"
LEFT = "LEFT JOIN"
RIGHT = "RIGHT JOIN"
FULL = "FULL OUTER JOIN"
CROSS = "CROSS JOIN"
class OrderDirection(Enum):
ASC = "ASC"
DESC = "DESC"
@dataclass
class SQLQuery:
"""SQL查询对象"""
statement: str
parameters: Tuple[Any, ...]
def execute_on(self, connection) -> Any:
"""在连接上执行查询"""
print(f"执行SQL: {self.statement}")
print(f"参数: {self.parameters}")
return None
class SQLBuilder:
"""
SQL查询建造者
特性:
- 类型安全的查询构建
- 参数化查询防止SQL注入
- 支持复杂查询(JOIN、子查询、聚合)
- 查询优化提示
"""
def __init__(self, table: str, alias: str = None):
self._main_table = table
self._main_alias = alias
self._select_fields: List[str] = ["*"]
self._distinct: bool = False
self._joins: List[str] = []
self._where_conditions: List[str] = []
self._where_params: List[Any] = []
self._group_by_fields: List[str] = []
self._having_conditions: List[str] = []
self._having_params: List[Any] = []
self._order_by: List[str] = []
self._limit: Optional[int] = None
self._offset: Optional[int] = None
self._lock_mode: Optional[str] = None
self._ctes: List[str] = []
def select(self, *fields: str) -> 'SQLBuilder':
"""指定查询字段"""
if fields:
self._select_fields = list(fields)
return self
def select_raw(self, expression: str) -> 'SQLBuilder':
"""添加原始表达式"""
self._select_fields.append(expression)
return self
def distinct(self) -> 'SQLBuilder':
"""去重"""
self._distinct = True
return self
def count(self, field: str = "*", alias: str = None) -> 'SQLBuilder':
"""计数"""
expr = f"COUNT({field})"
if alias:
expr += f" AS {alias}"
return self.select_raw(expr)
def sum(self, field: str, alias: str = None) -> 'SQLBuilder':
"""求和"""
expr = f"SUM({field})"
if alias:
expr += f" AS {alias}"
return self.select_raw(expr)
def avg(self, field: str, alias: str = None) -> 'SQLBuilder':
"""平均值"""
expr = f"AVG({field})"
if alias:
expr += f" AS {alias}"
return self.select_raw(expr)
def max(self, field: str, alias: str = None) -> 'SQLBuilder':
"""最大值"""
expr = f"MAX({field})"
if alias:
expr += f" AS {alias}"
return self.select_raw(expr)
def min(self, field: str, alias: str = None) -> 'SQLBuilder':
"""最小值"""
expr = f"MIN({field})"
if alias:
expr += f" AS {alias}"
return self.select_raw(expr)
def join(self, table: str, on: str,
join_type: JoinType = JoinType.INNER,
alias: str = None) -> 'SQLBuilder':
"""连接表"""
table_expr = f"{table}" + (f" AS {alias}" if alias else "")
self._joins.append(f"{join_type.value} {table_expr} ON {on}")
return self
def left_join(self, table: str, on: str, alias: str = None) -> 'SQLBuilder':
"""左连接"""
return self.join(table, on, JoinType.LEFT, alias)
def right_join(self, table: str, on: str, alias: str = None) -> 'SQLBuilder':
"""右连接"""
return self.join(table, on, JoinType.RIGHT, alias)
def where(self, condition: str, *params: Any) -> 'SQLBuilder':
"""添加WHERE条件"""
self._where_conditions.append(condition)
self._where_params.extend(params)
return self
def where_eq(self, field: str, value: Any) -> 'SQLBuilder':
"""等于条件"""
return self.where(f"{field} = ?", value)
def where_ne(self, field: str, value: Any) -> 'SQLBuilder':
"""不等于条件"""
return self.where(f"{field} != ?", value)
def where_gt(self, field: str, value: Any) -> 'SQLBuilder':
"""大于条件"""
return self.where(f"{field} > ?", value)
def where_gte(self, field: str, value: Any) -> 'SQLBuilder':
"""大于等于条件"""
return self.where(f"{field} >= ?", value)
def where_lt(self, field: str, value: Any) -> 'SQLBuilder':
"""小于条件"""
return self.where(f"{field} < ?", value)
def where_lte(self, field: str, value: Any) -> 'SQLBuilder':
"""小于等于条件"""
return self.where(f"{field} <= ?", value)
def where_like(self, field: str, pattern: str) -> 'SQLBuilder':
"""LIKE条件"""
return self.where(f"{field} LIKE ?", pattern)
def where_in(self, field: str, values: List[Any]) -> 'SQLBuilder':
"""IN条件"""
placeholders = ", ".join("?" * len(values))
return self.where(f"{field} IN ({placeholders})", *values)
def where_not_in(self, field: str, values: List[Any]) -> 'SQLBuilder':
"""NOT IN条件"""
placeholders = ", ".join("?" * len(values))
return self.where(f"{field} NOT IN ({placeholders})", *values)
def where_between(self, field: str, start: Any, end: Any) -> 'SQLBuilder':
"""BETWEEN条件"""
return self.where(f"{field} BETWEEN ? AND ?", start, end)
def where_null(self, field: str) -> 'SQLBuilder':
"""IS NULL条件"""
return self.where(f"{field} IS NULL")
def where_not_null(self, field: str) -> 'SQLBuilder':
"""IS NOT NULL条件"""
return self.where(f"{field} IS NOT NULL")
def where_exists(self, subquery: 'SQLBuilder') -> 'SQLBuilder':
"""EXISTS条件"""
sub_sql = subquery._build_select()
return self.where(f"EXISTS ({sub_sql.statement})", *sub_sql.parameters)
def group_by(self, *fields: str) -> 'SQLBuilder':
"""GROUP BY"""
self._group_by_fields.extend(fields)
return self
def having(self, condition: str, *params: Any) -> 'SQLBuilder':
"""HAVING条件"""
self._having_conditions.append(condition)
self._having_params.extend(params)
return self
def order_by(self, field: str, direction: OrderDirection = OrderDirection.ASC) -> 'SQLBuilder':
"""ORDER BY"""
self._order_by.append(f"{field} {direction.value}")
return self
def order_by_desc(self, field: str) -> 'SQLBuilder':
"""降序排序"""
return self.order_by(field, OrderDirection.DESC)
def limit(self, limit: int) -> 'SQLBuilder':
"""LIMIT"""
self._limit = limit
return self
def offset(self, offset: int) -> 'SQLBuilder':
"""OFFSET"""
self._offset = offset
return self
def page(self, page: int, page_size: int) -> 'SQLBuilder':
"""分页"""
self._limit = page_size
self._offset = (page - 1) * page_size
return self
def for_update(self) -> 'SQLBuilder':
"""行锁"""
self._lock_mode = "FOR UPDATE"
return self
def for_share(self) -> 'SQLBuilder':
"""共享锁"""
self._lock_mode = "FOR SHARE"
return self
def with_cte(self, name: str, query: 'SQLBuilder') -> 'SQLBuilder':
"""CTE (Common Table Expression)"""
sub_sql = query._build_select()
self._ctes.append(f"{name} AS ({sub_sql.statement})")
self._where_params[:0] = sub_sql.parameters
return self
def _build_select(self) -> SQLQuery:
"""构建SELECT语句"""
parts = []
params = []
if self._ctes:
parts.append(f"WITH {', '.join(self._ctes)}")
select_clause = "SELECT "
if self._distinct:
select_clause += "DISTINCT "
select_clause += ", ".join(self._select_fields)
parts.append(select_clause)
from_clause = f"FROM {self._main_table}"
if self._main_alias:
from_clause += f" AS {self._main_alias}"
parts.append(from_clause)
if self._joins:
parts.extend(self._joins)
if self._where_conditions:
parts.append(f"WHERE {' AND '.join(self._where_conditions)}")
params.extend(self._where_params)
if self._group_by_fields:
parts.append(f"GROUP BY {', '.join(self._group_by_fields)}")
if self._having_conditions:
parts.append(f"HAVING {' AND '.join(self._having_conditions)}")
params.extend(self._having_params)
if self._order_by:
parts.append(f"ORDER BY {', '.join(self._order_by)}")
if self._limit is not None:
parts.append(f"LIMIT {self._limit}")
if self._offset is not None:
parts.append(f"OFFSET {self._offset}")
if self._lock_mode:
parts.append(self._lock_mode)
return SQLQuery(
statement=" ".join(parts),
parameters=tuple(params)
)
def build(self) -> SQLQuery:
"""构建查询"""
return self._build_select()
query = (SQLBuilder("users", "u")
.select("u.id", "u.name", "u.email", "COUNT(o.id) AS order_count")
.left_join("orders o", "u.id = o.user_id")
.where_eq("u.status", "active")
.where_like("u.name", "%张%")
.where_gte("u.created_at", "2024-01-01")
.group_by("u.id", "u.name", "u.email")
.having("COUNT(o.id) > ?", 0)
.order_by_desc("order_count")
.page(1, 20)
.build())
print(f"SQL: {query.statement}")
print(f"参数: {query.parameters}")4.3.2 文档生成器
from typing import Optional, List, Dict, Any, Union
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
from enum import Enum, auto
class DocumentFormat(Enum):
MARKDOWN = "markdown"
HTML = "html"
PDF = "pdf"
DOCX = "docx"
@dataclass
class DocumentSection:
"""文档章节"""
title: str
level: int
content: str
subsections: List['DocumentSection'] = field(default_factory=list)
@dataclass
class Document:
"""文档"""
title: str
author: str = ""
date: str = ""
sections: List[DocumentSection] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def to_markdown(self) -> str:
"""转换为Markdown"""
lines = [f"# {self.title}\n"]
if self.author:
lines.append(f"**作者**: {self.author}\n")
if self.date:
lines.append(f"**日期**: {self.date}\n")
lines.append("\n---\n")
for section in self.sections:
lines.extend(self._section_to_markdown(section))
return "\n".join(lines)
def _section_to_markdown(self, section: DocumentSection, depth: int = 0) -> List[str]:
lines = []
prefix = "#" * (section.level + 1)
lines.append(f"\n{prefix} {section.title}\n")
if section.content:
lines.append(section.content)
for subsection in section.subsections:
lines.extend(self._section_to_markdown(subsection, depth + 1))
return lines
class DocumentBuilder:
"""
文档建造者
特性:
- 支持层级结构
- 多种输出格式
- 模板支持
"""
def __init__(self):
self._title: str = ""
self._author: str = ""
self._date: str = ""
self._sections: List[DocumentSection] = []
self._current_section_stack: List[DocumentSection] = []
self._metadata: Dict[str, Any] = {}
def title(self, title: str) -> 'DocumentBuilder':
self._title = title
return self
def author(self, author: str) -> 'DocumentBuilder':
self._author = author
return self
def date(self, date: str) -> 'DocumentBuilder':
self._date = date
return self
def metadata(self, key: str, value: Any) -> 'DocumentBuilder':
self._metadata[key] = value
return self
def section(self, title: str, content: str = "") -> 'DocumentBuilder':
"""添加章节"""
section = DocumentSection(
title=title,
level=len(self._current_section_stack) + 1,
content=content
)
if self._current_section_stack:
self._current_section_stack[-1].subsections.append(section)
else:
self._sections.append(section)
return self
def begin_section(self, title: str) -> 'DocumentBuilder':
"""开始章节"""
section = DocumentSection(
title=title,
level=len(self._current_section_stack) + 1,
content=""
)
if self._current_section_stack:
self._current_section_stack[-1].subsections.append(section)
else:
self._sections.append(section)
self._current_section_stack.append(section)
return self
def end_section(self) -> 'DocumentBuilder':
"""结束章节"""
if self._current_section_stack:
self._current_section_stack.pop()
return self
def paragraph(self, text: str) -> 'DocumentBuilder':
"""添加段落"""
if self._current_section_stack:
current = self._current_section_stack[-1]
if current.content:
current.content += "\n\n" + text
else:
current.content = text
return self
def code_block(self, code: str, language: str = "") -> 'DocumentBuilder':
"""添加代码块"""
block = f"\n\n```{language}\n{code}\n```\n"
if self._current_section_stack:
current = self._current_section_stack[-1]
current.content += block
return self
def list_items(self, *items: str) -> 'DocumentBuilder':
"""添加列表"""
list_text = "\n\n" + "\n".join(f"- {item}" for item in items) + "\n"
if self._current_section_stack:
current = self._current_section_stack[-1]
current.content += list_text
return self
def build(self) -> Document:
if not self._title:
raise ValueError("文档标题不能为空")
return Document(
title=self._title,
author=self._author,
date=self._date,
sections=self._sections,
metadata=self._metadata
)
doc = (DocumentBuilder()
.title("Python设计模式指南")
.author("张三")
.date("2024-01-15")
.metadata("version", "1.0")
.metadata("category", "技术文档")
.begin_section("概述")
.paragraph("设计模式是软件工程中的重要概念,它提供了可复用的解决方案。")
.paragraph("本指南将介绍23种经典设计模式及其Python实现。")
.end_section()
.begin_section("创建型模式")
.paragraph("创建型模式关注对象的创建过程。")
.list_items("单例模式", "工厂方法模式", "抽象工厂模式", "建造者模式", "原型模式")
.begin_section("单例模式")
.paragraph("单例模式确保一个类只有一个实例。")
.code_block("""
class Singleton:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
""", "python")
.end_section()
.end_section()
.build())
print(doc.to_markdown())4.4 建造者模式 vs 工厂模式
4.4.1 核心区别
┌─────────────────────────────────────────────────────────────────┐
│ 建造者模式 vs 工厂模式 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 工厂模式 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 关注点: 创建什么(What) │ │
│ │ 方式: 一次性创建完整对象 │ │
│ │ 参数: 通过构造函数参数传递 │ │
│ │ 适用: 对象类型不确定,但创建过程简单 │ │
│ │ 示例: DatabaseFactory.create_connection("mysql") │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
│ 建造者模式 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 关注点: 如何创建(How) │ │
│ │ 方式: 分步骤构建对象 │ │
│ │ 参数: 通过方法调用设置 │ │
│ │ 适用: 创建过程复杂,有多个可选参数 │ │
│ │ 示例: ComputerBuilder().set_cpu("i9").set_ram("64GB") │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘4.4.2 模式协作
建造者模式可以与工厂模式协作:
class ComputerFactory:
"""计算机工厂:根据用途选择合适的建造者"""
@staticmethod
def create_gaming_computer() -> Computer:
builder = GamingComputerBuilder()
director = ComputerDirector(builder)
return director.build_high_end_gaming()
@staticmethod
def create_office_computer() -> Computer:
builder = OfficeComputerBuilder()
director = ComputerDirector(builder)
return director.build_office_workstation()
@staticmethod
def create_custom_computer(builder: ComputerBuilder) -> Computer:
return builder.get_result()4.4.3 选择决策表
| 场景特征 | 推荐模式 |
|---|---|
| 对象创建过程简单 | 工厂模式 |
| 对象有多个可选参数 | 建造者模式 |
| 需要分步骤构建 | 建造者模式 |
| 需要验证构建约束 | 建造者模式 |
| 需要构建不可变对象 | 建造者模式 |
| 需要支持多种表示 | 建造者模式 + Director |
| 对象类型运行时决定 | 工厂模式 |
4.5 反模式与最佳实践
4.5.1 常见反模式
反模式1:建造者过度使用
class SimpleObject:
"""简单对象不需要建造者"""
def __init__(self, name: str, value: int):
self.name = name
self.value = value
class SimpleObjectBuilder:
"""过度设计:简单对象不需要建造者"""
def name(self, name: str) -> 'SimpleObjectBuilder':
self._name = name
return self
def value(self, value: int) -> 'SimpleObjectBuilder':
self._value = value
return self
def build(self) -> SimpleObject:
return SimpleObject(self._name, self._value)
obj = SimpleObject("test", 42)反模式2:建造者状态污染
class BadBuilder:
"""错误示例:建造者持有可变状态"""
def __init__(self):
self._product = Product()
def build(self) -> Product:
return self._product
builder = BadBuilder()
builder.build()
builder.build()正确做法:
class GoodBuilder:
"""正确示例:每次构建创建新实例"""
def __init__(self):
self._reset()
def _reset(self):
self._product = Product()
def build(self) -> Product:
result = self._product
self._reset()
return result4.5.2 最佳实践
from typing import TypeVar, Generic, Callable, List, Any
T = TypeVar('T')
class BuilderPattern:
"""
建造者模式最佳实践
原则:
1. 建造者应该创建独立的产品实例
2. 提供合理的默认值
3. 在build()时验证约束
4. 支持链式调用
5. 考虑不可变性
"""
pass
class BestPracticeBuilder(Generic[T]):
"""最佳实践建造者模板"""
def __init__(self):
self._validators: List[Callable[[T], bool]] = []
self._transformers: List[Callable[[T], T]] = []
def validate(self, validator: Callable[[T], bool]) -> 'BestPracticeBuilder':
self._validators.append(validator)
return self
def transform(self, transformer: Callable[[T], T]) -> 'BestPracticeBuilder':
self._transformers.append(transformer)
return self
def _validate_product(self, product: T) -> None:
for validator in self._validators:
if not validator(product):
raise ValueError("产品验证失败")
def _transform_product(self, product: T) -> T:
for transformer in self._transformers:
product = transformer(product)
return product4.6 模式评估与决策指南
4.6.1 适用性检查清单
| 检查项 | 是 | 否 | 说明 |
|---|---|---|---|
| 对象有多个可选参数? | ✓ 使用 | ✗ 评估 | 参数复杂性 |
| 构建过程需要验证? | ✓ 使用 | ✗ 评估 | 约束检查 |
| 需要分步构建? | ✓ 使用 | ✗ 评估 | 构建流程 |
| 需要不可变对象? | ✓ 使用 | ✗ 评估 | 安全性需求 |
| 需要多种表示? | ✓ 使用 | ✗ 评估 | 变体支持 |
4.6.2 实现方式对比
| 实现方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 经典建造者 | 结构清晰、支持Director | 代码量大 | 标准实现 |
| 流畅建造者 | API简洁、链式调用 | 无Director | 简单场景 |
| 泛型建造者 | 类型安全、可复用 | 学习曲线 | 框架开发 |
| 不可变建造者 | 线程安全、缓存优化 | 修改需重建 | 并发场景 |
4.7 小结
建造者模式通过分步构建解决了复杂对象的创建问题:
┌─────────────────────────────────────────────────────────────────┐
│ 建造者模式决策树 │
│ │
│ 对象创建是否复杂? │
│ │ │
│ ┌──────────┴──────────┐ │
│ ▼ ▼ │
│ 是的 不复杂 │
│ │ │ │
│ ▼ ▼ │
│ 需要多种表示? 直接构造函数 │
│ │ │
│ ┌───────┴───────┐ │
│ ▼ ▼ │
│ 是 否 │
│ │ │ │
│ ▼ ▼ │
│ 建造者+Director 流畅建造者 │
│ │
└─────────────────────────────────────────────────────────────────┘关键要点:
- 核心价值:分离构建与表示,支持分步构建
- 链式调用:提供流畅的API体验
- 验证机制:在构建时验证约束条件
- 不可变性:支持构建不可变对象
- Director角色:封装标准构建流程(可选)
在下一章中,我们将探讨原型模式,学习如何通过克隆创建对象。