第10章 外观模式
学习目标
完成本章学习后，读者将能够：
- 理解外观模式的核心概念与数学形式化定义
- 掌握简化复杂系统接口的设计方法
- 运用外观模式降低系统耦合度
- 实现多层外观与可扩展外观模式
- 识别外观模式的适用场景与反模式
- 设计企业级外观模式解决方案
10.1 模式定义
10.1.1 核心定义
外观模式（Facade Pattern） 为子系统中的一组接口提供一个一致的界面，外观模式定义了一个高层接口，这个接口使得这一子系统更加容易使用。外观模式是一种结构型模式，它通过封装复杂的子系统，为客户端提供一个简化的调用入口。
10.1.2 形式化定义
设 $\mathcal{S} = {S_1, S_2, ..., S_n}$ 为子系统集合，每个子系统 $S_i$ 提供操作集合 $O_i = {o_{i1}, o_{i2}, ..., o_{im}}$。外观模式可形式化定义为：
$$\text{Facade}: \mathcal{F} \rightarrow \bigcup_{i=1}^{n} O_i$$
其中 $\mathcal{F}$ 为外观接口，满足：
$$\forall f \in \mathcal{F}: f = \phi(o_{1j}, o_{2k}, ..., o_{nl})$$
$\phi$ 为操作组合函数，将多个子系统操作封装为单一外观操作。
接口简化度：
$$\text{Simplification} = \frac{|\mathcal{F}|}{|\bigcup_{i=1}^{n} O_i|}$$
耦合度降低：
$$\text{Coupling}{\text{with_facade}} = \frac{\text{Client} \rightarrow \text{Facade}}{\text{Client} \rightarrow \sum^{n} S_i}$$
迪米特法则满足度：
$$\text{LoD Compliance} = \begin{cases} 1 & \text{if Client only knows Facade} \ 0 & \text{otherwise} \end{cases}$$
复杂度分析：
- 外观创建：$O(n)$（$n$ 为子系统数量）
- 操作调用：$O(1)$（单一入口点）
- 空间复杂度：$O(n)$（持有子系统引用）
10.1.3 历史背景与学术脉络
外观模式起源于软件架构分层的设计实践。其学术发展历程如下：
| 年份 | 里程碑 | 贡献者 |
|---|---|---|
| 1970s | 分层架构概念提出 | Dijkstra |
| 1987 | ET++框架中的外观应用 | Weinand, Gamma |
| 1994 | GoF《设计模式》正式收录 | Gang of Four |
| 1996 | 与迪米特法则关联研究 | Lieberherr |
| 2002 | API设计中的外观应用 | Bloch |
| 2010 | 微服务外观模式研究 | Richardson |
| 2018 | API网关作为外观模式 | 云原生社区 |
学术意义：外观模式是**迪米特法则（LoD）**的典型应用，体现了"最少知识原则"的设计思想，为复杂系统的简化访问提供了范式解决方案。在微服务架构中，API网关模式是外观模式的现代演进。
10.2 模式结构与参与者
10.2.1 UML类图
┌─────────────────────────────────────────────────────────────────────────┐
│ Client │
└───────────────────────────────────┬─────────────────────────────────────┘
│ uses
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Facade │
├─────────────────────────────────────────────────────────────────────────┤
│ - subsystemA: SubsystemA │
│ - subsystemB: SubsystemB │
│ - subsystemC: SubsystemC │
├─────────────────────────────────────────────────────────────────────────┤
│ + operation1(): Result │
│ + operation2(): Result │
│ + operation3(): Result │
└───────────────────────────────────┬─────────────────────────────────────┘
│ delegates to
┌───────────────────────────┼───────────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ SubsystemA │ │ SubsystemB │ │ SubsystemC │
├───────────────┤ ├───────────────┤ ├───────────────┤
│ + methodA1() │ │ + methodB1() │ │ + methodC1() │
│ + methodA2() │ │ + methodB2() │ │ + methodC2() │
│ + methodA3() │ │ + methodB3() │ │ + methodC3() │
└───────────────┘ └───────────────┘ └───────────────┘10.2.2 参与者职责
| 参与者 | 职责 | 关键特征 |
|---|---|---|
| Facade | 提供简化接口，委托子系统操作 | 知道哪些子系统负责处理请求 |
| Subsystem | 实现专业功能，处理外观委派的任务 | 不知道外观的存在 |
| Client | 通过外观与子系统交互 | 不直接访问子系统 |
10.2.3 调用流程
Client Facade SubsystemA SubsystemB SubsystemC
│ │ │ │ │
│──── operation1() ──────>│ │ │ │
│ │──── methodA1() ───────>│ │ │
│ │<─── result A ──────────│ │ │
│ │──── methodB1() ────────────────────>│ │
│ │<─── result B ────────────────────────────────────│
│ │──── methodC1() ────────────────────────────────────────────>│
│ │<─── result C ──────────────────────────────────────────────│
│<─── combined result ────│ │ │ │
│ │ │ │ │10.3 Python实现
10.3.1 标准实现
python
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
class SubsystemA:
"""子系统A - 负责数据处理"""
def operation_a1(self) -> str:
return "SubsystemA: 数据验证完成"
def operation_a2(self) -> str:
return "SubsystemA: 数据转换完成"
def process_data(self, data: Any) -> Dict[str, Any]:
return {"processed": True, "data": data}
class SubsystemB:
"""子系统B - 负责业务逻辑"""
def operation_b1(self) -> str:
return "SubsystemB: 业务规则检查完成"
def operation_b2(self) -> str:
return "SubsystemB: 业务计算完成"
def apply_business_rules(self, data: Dict) -> Dict:
data["business_applied"] = True
return data
class SubsystemC:
"""子系统C - 负责存储"""
def operation_c1(self) -> str:
return "SubsystemC: 存储准备完成"
def operation_c2(self) -> str:
return "SubsystemC: 数据持久化完成"
def save(self, data: Dict) -> bool:
print(f"保存数据: {data}")
return True
class Facade:
"""外观类 - 提供简化接口"""
def __init__(self):
self._subsystem_a = SubsystemA()
self._subsystem_b = SubsystemB()
self._subsystem_c = SubsystemC()
def operation1(self) -> str:
"""简化操作1 - 初始化流程"""
results = [
"Facade: 初始化子系统",
self._subsystem_a.operation_a1(),
self._subsystem_b.operation_b1(),
self._subsystem_c.operation_c1(),
]
return "\n".join(results)
def operation2(self) -> str:
"""简化操作2 - 执行流程"""
results = [
"Facade: 执行复杂操作",
self._subsystem_a.operation_a2(),
self._subsystem_b.operation_b2(),
self._subsystem_c.operation_c2(),
]
return "\n".join(results)
def process_complete(self, data: Any) -> Dict[str, Any]:
"""完整处理流程"""
result = self._subsystem_a.process_data(data)
result = self._subsystem_b.apply_business_rules(result)
success = self._subsystem_c.save(result)
return {"success": success, "result": result}
def client_code(facade: Facade) -> None:
"""客户端代码 - 仅与外观交互"""
print(facade.operation1())
print()
print(facade.operation2())
print()
print(facade.process_complete({"input": "test"}))
if __name__ == "__main__":
facade = Facade()
client_code(facade)10.3.2 Protocol实现（结构化类型）
python
from typing import Protocol, runtime_checkable, Dict, Any, List, Optional
from dataclasses import dataclass
@runtime_checkable
class SubsystemProtocol(Protocol):
"""子系统协议"""
def initialize(self) -> bool: ...
def process(self, data: Dict[str, Any]) -> Dict[str, Any]: ...
def cleanup(self) -> None: ...
@dataclass
class ValidationSubsystem:
"""验证子系统"""
def initialize(self) -> bool:
print("验证子系统初始化")
return True
def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
if not data.get("value"):
raise ValueError("缺少必要字段")
data["validated"] = True
return data
def cleanup(self) -> None:
print("验证子系统清理")
@dataclass
class TransformSubsystem:
"""转换子系统"""
def initialize(self) -> bool:
print("转换子系统初始化")
return True
def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
data["transformed"] = True
data["value"] = str(data.get("value", "")).upper()
return data
def cleanup(self) -> None:
print("转换子系统清理")
@dataclass
class StorageSubsystem:
"""存储子系统"""
def initialize(self) -> bool:
print("存储子系统初始化")
return True
def process(self, data: Dict[str, Any]) -> Dict[str, Any]:
data["stored"] = True
print(f"存储数据: {data}")
return data
def cleanup(self) -> None:
print("存储子系统清理")
class ModularFacade:
"""模块化外观 - 支持动态子系统"""
def __init__(self):
self._subsystems: List[SubsystemProtocol] = []
def register_subsystem(self, subsystem: SubsystemProtocol) -> None:
self._subsystems.append(subsystem)
def initialize_all(self) -> bool:
results = [s.initialize() for s in self._subsystems]
return all(results)
def process_pipeline(self, data: Dict[str, Any]) -> Dict[str, Any]:
result = data.copy()
for subsystem in self._subsystems:
result = subsystem.process(result)
return result
def cleanup_all(self) -> None:
for subsystem in self._subsystems:
subsystem.cleanup()
if __name__ == "__main__":
facade = ModularFacade()
facade.register_subsystem(ValidationSubsystem())
facade.register_subsystem(TransformSubsystem())
facade.register_subsystem(StorageSubsystem())
facade.initialize_all()
result = facade.process_pipeline({"value": "hello world"})
print(f"\n处理结果: {result}")
facade.cleanup_all()10.3.3 泛型实现（类型安全）
python
from abc import ABC, abstractmethod
from typing import TypeVar, Generic, Dict, Any, List, Optional, Callable
from dataclasses import dataclass
T = TypeVar('T')
R = TypeVar('R')
class SubsystemBase(ABC, Generic[T, R]):
"""子系统基类"""
@abstractmethod
def process(self, input_data: T) -> R:
pass
@abstractmethod
def get_name(self) -> str:
pass
class GenericFacade(Generic[T, R]):
"""泛型外观"""
def __init__(self):
self._subsystems: List[SubsystemBase] = []
self._preprocessors: List[Callable] = []
self._postprocessors: List[Callable] = []
def add_subsystem(self, subsystem: SubsystemBase) -> 'GenericFacade':
self._subsystems.append(subsystem)
return self
def add_preprocessor(self, processor: Callable) -> 'GenericFacade':
self._preprocessors.append(processor)
return self
def add_postprocessor(self, processor: Callable) -> 'GenericFacade':
self._postprocessors.append(processor)
return self
def execute(self, input_data: T) -> R:
data = input_data
for preprocessor in self._preprocessors:
data = preprocessor(data)
for subsystem in self._subsystems:
data = subsystem.process(data)
for postprocessor in self._postprocessors:
data = postprocessor(data)
return data
@dataclass
class StringValidationSubsystem(SubsystemBase[str, str]):
"""字符串验证子系统"""
def process(self, input_data: str) -> str:
if not input_data:
raise ValueError("输入不能为空")
return input_data.strip()
def get_name(self) -> str:
return "StringValidation"
@dataclass
class StringTransformSubsystem(SubsystemBase[str, str]):
"""字符串转换子系统"""
def process(self, input_data: str) -> str:
return input_data.upper()
def get_name(self) -> str:
return "StringTransform"
@dataclass
class StringFormattingSubsystem(SubsystemBase[str, Dict[str, Any]]):
"""字符串格式化子系统"""
def process(self, input_data: str) -> Dict[str, Any]:
return {
"original": input_data,
"length": len(input_data),
"words": len(input_data.split()),
"formatted": f"[{input_data}]"
}
def get_name(self) -> str:
return "StringFormatting"
if __name__ == "__main__":
facade: GenericFacade[str, Dict[str, Any]] = GenericFacade()
facade.add_subsystem(StringValidationSubsystem())
facade.add_subsystem(StringTransformSubsystem())
facade.add_subsystem(StringFormattingSubsystem())
result = facade.execute(" hello world ")
print(f"处理结果: {result}")10.3.4 可配置外观
python
from typing import Any, Dict, List, Optional, Callable, Type
from dataclasses import dataclass, field
from enum import Enum
class OperationMode(Enum):
STRICT = "strict"
LENIENT = "lenient"
DEBUG = "debug"
@dataclass
class FacadeConfig:
"""外观配置"""
mode: OperationMode = OperationMode.STRICT
enable_logging: bool = True
enable_caching: bool = False
max_retries: int = 3
timeout: float = 30.0
class ConfigurableFacade:
"""可配置外观"""
def __init__(self, config: FacadeConfig = None):
self._config = config or FacadeConfig()
self._subsystems: Dict[str, Any] = {}
self._operations: Dict[str, Callable] = {}
self._cache: Dict[str, Any] = {}
def register_subsystem(self, name: str, subsystem: Any) -> None:
self._subsystems[name] = subsystem
def register_operation(
self,
name: str,
operation: Callable,
subsystems: List[str] = None
) -> None:
self._operations[name] = {
"handler": operation,
"subsystems": subsystems or []
}
def execute(self, operation_name: str, *args, **kwargs) -> Any:
if operation_name not in self._operations:
raise ValueError(f"未知操作: {operation_name}")
op_config = self._operations[operation_name]
if self._config.enable_logging:
print(f"[LOG] 执行操作: {operation_name}")
if self._config.enable_caching:
cache_key = f"{operation_name}:{args}:{kwargs}"
if cache_key in self._cache:
return self._cache[cache_key]
try:
result = op_config["handler"](*args, **kwargs)
if self._config.enable_caching:
self._cache[cache_key] = result
return result
except Exception as e:
if self._config.mode == OperationMode.STRICT:
raise
elif self._config.mode == OperationMode.LENIENT:
print(f"[ERROR] 操作失败: {e}")
return None
elif self._config.mode == OperationMode.DEBUG:
import traceback
traceback.print_exc()
raise
def get_subsystem(self, name: str) -> Optional[Any]:
return self._subsystems.get(name)
def list_operations(self) -> List[str]:
return list(self._operations.keys())
def clear_cache(self) -> None:
self._cache.clear()
if __name__ == "__main__":
class MockDatabase:
def query(self, sql: str) -> List[Dict]:
return [{"id": 1, "name": "test"}]
class MockCache:
def get(self, key: str) -> Optional[Any]:
return None
def set(self, key: str, value: Any) -> None:
pass
config = FacadeConfig(
mode=OperationMode.DEBUG,
enable_logging=True,
enable_caching=True
)
facade = ConfigurableFacade(config)
facade.register_subsystem("database", MockDatabase())
facade.register_subsystem("cache", MockCache())
facade.register_operation(
"get_users",
lambda: facade.get_subsystem("database").query("SELECT * FROM users")
)
print(f"可用操作: {facade.list_operations()}")
result = facade.execute("get_users")
print(f"查询结果: {result}")10.4 实际应用示例
10.4.1 电商订单处理系统
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
from enum import Enum
import uuid
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
@dataclass
class Product:
id: str
name: str
price: float
stock: int
@dataclass
class OrderItem:
product_id: str
product_name: str
quantity: int
unit_price: float
@property
def total(self) -> float:
return self.quantity * self.unit_price
@dataclass
class Order:
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
customer_id: str = ""
items: List[OrderItem] = field(default_factory=list)
shipping_address: str = ""
status: OrderStatus = OrderStatus.PENDING
created_at: datetime = field(default_factory=datetime.now)
payment_id: Optional[str] = None
shipment_id: Optional[str] = None
@property
def total_amount(self) -> float:
return sum(item.total for item in self.items)
class InventoryService:
"""库存服务"""
def __init__(self):
self._inventory: Dict[str, int] = {}
def set_stock(self, product_id: str, quantity: int) -> None:
self._inventory[product_id] = quantity
def check_availability(self, product_id: str, quantity: int) -> bool:
available = self._inventory.get(product_id, 0)
return available >= quantity
def reserve(self, product_id: str, quantity: int) -> bool:
if not self.check_availability(product_id, quantity):
return False
self._inventory[product_id] -= quantity
return True
def release(self, product_id: str, quantity: int) -> None:
self._inventory[product_id] = self._inventory.get(product_id, 0) + quantity
class PaymentService:
"""支付服务"""
def __init__(self):
self._transactions: Dict[str, Dict] = {}
def process_payment(
self,
order_id: str,
amount: float,
method: str
) -> str:
transaction_id = f"TXN-{uuid.uuid4().hex[:8]}"
self._transactions[transaction_id] = {
"order_id": order_id,
"amount": amount,
"method": method,
"status": "completed",
"created_at": datetime.now()
}
print(f"支付处理: {transaction_id}, 金额: ¥{amount:.2f}")
return transaction_id
def refund(self, transaction_id: str) -> bool:
if transaction_id in self._transactions:
self._transactions[transaction_id]["status"] = "refunded"
print(f"退款处理: {transaction_id}")
return True
return False
class ShippingService:
"""物流服务"""
def __init__(self):
self._shipments: Dict[str, Dict] = {}
def create_shipment(
self,
order_id: str,
address: str
) -> str:
shipment_id = f"SHP-{uuid.uuid4().hex[:8]}"
self._shipments[shipment_id] = {
"order_id": order_id,
"address": address,
"status": "pending",
"created_at": datetime.now()
}
print(f"创建发货单: {shipment_id}, 地址: {address}")
return shipment_id
def track(self, shipment_id: str) -> Dict[str, Any]:
return self._shipments.get(shipment_id, {})
class NotificationService:
"""通知服务"""
def send_order_confirmation(
self,
email: str,
order_id: str
) -> None:
print(f"发送订单确认邮件到 {email}: 订单 {order_id}")
def send_shipping_notification(
self,
email: str,
shipment_id: str
) -> None:
print(f"发送发货通知到 {email}: 发货单 {shipment_id}")
def send_sms(self, phone: str, message: str) -> None:
print(f"发送短信到 {phone}: {message}")
class LoyaltyService:
"""会员积分服务"""
def __init__(self):
self._points: Dict[str, int] = {}
def earn_points(
self,
customer_id: str,
amount: float
) -> int:
points = int(amount / 10)
self._points[customer_id] = self._points.get(customer_id, 0) + points
print(f"积分入账: 客户 {customer_id}, +{points} 积分")
return points
def get_points(self, customer_id: str) -> int:
return self._points.get(customer_id, 0)
class OrderFacade:
"""订单外观 - 统一订单处理接口"""
def __init__(self):
self._inventory = InventoryService()
self._payment = PaymentService()
self._shipping = ShippingService()
self._notification = NotificationService()
self._loyalty = LoyaltyService()
self._orders: Dict[str, Order] = {}
def create_order(
self,
customer_id: str,
items: List[Dict[str, Any]],
shipping_address: str
) -> Order:
"""创建订单"""
order = Order(
customer_id=customer_id,
shipping_address=shipping_address
)
for item in items:
order_item = OrderItem(
product_id=item["product_id"],
product_name=item["product_name"],
quantity=item["quantity"],
unit_price=item["unit_price"]
)
order.items.append(order_item)
self._orders[order.id] = order
print(f"创建订单: {order.id}, 客户: {customer_id}")
return order
def confirm_order(self, order_id: str) -> bool:
"""确认订单 - 检查并预留库存"""
order = self._orders.get(order_id)
if not order:
return False
reserved_items = []
try:
for item in order.items:
if not self._inventory.reserve(item.product_id, item.quantity):
raise ValueError(f"库存不足: {item.product_id}")
reserved_items.append(item)
order.status = OrderStatus.CONFIRMED
print(f"订单确认: {order_id}")
return True
except ValueError as e:
for item in reserved_items:
self._inventory.release(item.product_id, item.quantity)
print(f"订单确认失败: {e}")
return False
def process_payment(
self,
order_id: str,
payment_method: str
) -> bool:
"""处理支付"""
order = self._orders.get(order_id)
if not order or order.status != OrderStatus.CONFIRMED:
return False
try:
transaction_id = self._payment.process_payment(
order_id,
order.total_amount,
payment_method
)
order.payment_id = transaction_id
order.status = OrderStatus.PAID
print(f"支付完成: {order_id}")
return True
except Exception as e:
print(f"支付失败: {e}")
return False
def ship_order(
self,
order_id: str,
customer_email: str
) -> bool:
"""发货"""
order = self._orders.get(order_id)
if not order or order.status != OrderStatus.PAID:
return False
shipment_id = self._shipping.create_shipment(
order_id,
order.shipping_address
)
order.shipment_id = shipment_id
order.status = OrderStatus.SHIPPED
self._notification.send_shipping_notification(customer_email, shipment_id)
self._loyalty.earn_points(order.customer_id, order.total_amount)
print(f"发货完成: {order_id}")
return True
def place_order(
self,
customer_id: str,
items: List[Dict[str, Any]],
shipping_address: str,
payment_method: str,
customer_email: str
) -> Dict[str, Any]:
"""一键下单 - 完整流程"""
print("\n=== 开始处理订单 ===")
order = self.create_order(customer_id, items, shipping_address)
if not self.confirm_order(order.id):
return {"success": False, "error": "库存不足"}
if not self.process_payment(order.id, payment_method):
self.cancel_order(order.id)
return {"success": False, "error": "支付失败"}
self._notification.send_order_confirmation(customer_email, order.id)
if not self.ship_order(order.id, customer_email):
return {"success": False, "error": "发货失败"}
print("=== 订单处理完成 ===\n")
return {
"success": True,
"order_id": order.id,
"payment_id": order.payment_id,
"shipment_id": order.shipment_id,
"total_amount": order.total_amount
}
def cancel_order(self, order_id: str) -> bool:
"""取消订单"""
order = self._orders.get(order_id)
if not order:
return False
if order.status in [OrderStatus.SHIPPED, OrderStatus.DELIVERED]:
print("无法取消已发货或已送达的订单")
return False
for item in order.items:
self._inventory.release(item.product_id, item.quantity)
if order.payment_id:
self._payment.refund(order.payment_id)
order.status = OrderStatus.CANCELLED
print(f"订单已取消: {order_id}")
return True
def get_order(self, order_id: str) -> Optional[Order]:
return self._orders.get(order_id)
def set_product_stock(self, product_id: str, quantity: int) -> None:
self._inventory.set_stock(product_id, quantity)
if __name__ == "__main__":
facade = OrderFacade()
facade.set_product_stock("PROD-001", 100)
facade.set_product_stock("PROD-002", 50)
result = facade.place_order(
customer_id="CUST-001",
items=[
{"product_id": "PROD-001", "product_name": "Python教程", "quantity": 2, "unit_price": 99.0},
{"product_id": "PROD-002", "product_name": "设计模式书籍", "quantity": 1, "unit_price": 128.0}
],
shipping_address="北京市朝阳区xxx街道",
payment_method="credit_card",
customer_email="customer@example.com"
)
print(f"\n订单结果: {result}")10.4.2 数据库操作外观
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, TypeVar, Generic, Callable
from contextlib import contextmanager
from enum import Enum
import time
T = TypeVar('T')
class ConnectionStatus(Enum):
DISCONNECTED = "disconnected"
CONNECTED = "connected"
ERROR = "error"
@dataclass
class ConnectionConfig:
host: str
port: int
database: str
username: str
password: str
pool_size: int = 5
timeout: float = 30.0
@dataclass
class QueryResult:
success: bool
data: List[Dict[str, Any]] = field(default_factory=list)
affected_rows: int = 0
error: Optional[str] = None
execution_time: float = 0.0
class ConnectionPool:
"""连接池"""
def __init__(self, config: ConnectionConfig):
self._config = config
self._connections: List[Any] = []
self._available: List[Any] = []
self._status = ConnectionStatus.DISCONNECTED
def initialize(self) -> bool:
print(f"初始化连接池: {self._config.host}:{self._config.port}")
for _ in range(self._config.pool_size):
conn = self._create_connection()
self._connections.append(conn)
self._available.append(conn)
self._status = ConnectionStatus.CONNECTED
return True
def _create_connection(self) -> Any:
return {"id": id(self), "created_at": time.time()}
def acquire(self) -> Any:
if not self._available:
raise RuntimeError("连接池已满")
return self._available.pop()
def release(self, conn: Any) -> None:
self._available.append(conn)
def close_all(self) -> None:
self._connections.clear()
self._available.clear()
self._status = ConnectionStatus.DISCONNECTED
@property
def status(self) -> ConnectionStatus:
return self._status
class QueryBuilder:
"""查询构建器"""
def select(
self,
table: str,
columns: List[str] = None,
where: Dict[str, Any] = None,
order_by: str = None,
limit: int = None
) -> str:
cols = ", ".join(columns) if columns else "*"
sql = f"SELECT {cols} FROM {table}"
if where:
conditions = [f"{k} = %s" for k in where.keys()]
sql += f" WHERE {' AND '.join(conditions)}"
if order_by:
sql += f" ORDER BY {order_by}"
if limit:
sql += f" LIMIT {limit}"
return sql
def insert(self, table: str, data: Dict[str, Any]) -> str:
columns = ", ".join(data.keys())
placeholders = ", ".join(["%s"] * len(data))
return f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
def update(
self,
table: str,
data: Dict[str, Any],
where: Dict[str, Any]
) -> str:
sets = ", ".join([f"{k} = %s" for k in data.keys()])
conditions = " AND ".join([f"{k} = %s" for k in where.keys()])
return f"UPDATE {table} SET {sets} WHERE {conditions}"
def delete(self, table: str, where: Dict[str, Any]) -> str:
conditions = " AND ".join([f"{k} = %s" for k in where.keys()])
return f"DELETE FROM {table} WHERE {conditions}"
class QueryExecutor:
"""查询执行器"""
def execute(
self,
conn: Any,
sql: str,
params: tuple = ()
) -> QueryResult:
start_time = time.time()
try:
print(f"执行SQL: {sql}")
if params:
print(f"参数: {params}")
mock_data = [{"id": i, "name": f"record_{i}"} for i in range(1, 4)]
execution_time = time.time() - start_time
return QueryResult(
success=True,
data=mock_data,
affected_rows=len(mock_data),
execution_time=execution_time
)
except Exception as e:
return QueryResult(
success=False,
error=str(e),
execution_time=time.time() - start_time
)
class TransactionManager:
"""事务管理器"""
def __init__(self, executor: QueryExecutor, pool: ConnectionPool):
self._executor = executor
self._pool = pool
self._connection = None
self._active = False
def begin(self) -> None:
self._connection = self._pool.acquire()
self._active = True
print("BEGIN TRANSACTION")
def commit(self) -> None:
print("COMMIT")
self._pool.release(self._connection)
self._active = False
def rollback(self) -> None:
print("ROLLBACK")
self._pool.release(self._connection)
self._active = False
def execute(self, sql: str, params: tuple = ()) -> QueryResult:
if not self._active:
raise RuntimeError("事务未激活")
return self._executor.execute(self._connection, sql, params)
class CacheLayer:
"""缓存层"""
def __init__(self):
self._cache: Dict[str, Any] = {}
self._ttl: Dict[str, float] = {}
def get(self, key: str) -> Optional[Any]:
if key in self._cache:
if time.time() < self._ttl.get(key, 0):
return self._cache[key]
del self._cache[key]
del self._ttl[key]
return None
def set(self, key: str, value: Any, ttl: float = 300) -> None:
self._cache[key] = value
self._ttl[key] = time.time() + ttl
def delete(self, key: str) -> None:
self._cache.pop(key, None)
self._ttl.pop(key, None)
def clear(self) -> None:
self._cache.clear()
self._ttl.clear()
class DatabaseFacade:
"""数据库外观 - 统一数据库操作接口"""
def __init__(self, config: ConnectionConfig):
self._pool = ConnectionPool(config)
self._query_builder = QueryBuilder()
self._executor = QueryExecutor()
self._cache = CacheLayer()
self._transaction: Optional[TransactionManager] = None
def connect(self) -> bool:
return self._pool.initialize()
def disconnect(self) -> None:
self._pool.close_all()
self._cache.clear()
def find_all(
self,
table: str,
columns: List[str] = None,
where: Dict[str, Any] = None,
use_cache: bool = True
) -> QueryResult:
cache_key = f"find_all:{table}:{columns}:{where}"
if use_cache:
cached = self._cache.get(cache_key)
if cached:
return QueryResult(success=True, data=cached)
sql = self._query_builder.select(table, columns, where)
params = tuple(where.values()) if where else ()
conn = self._pool.acquire()
try:
result = self._executor.execute(conn, sql, params)
if result.success and use_cache:
self._cache.set(cache_key, result.data)
return result
finally:
self._pool.release(conn)
def find_one(
self,
table: str,
where: Dict[str, Any],
columns: List[str] = None
) -> QueryResult:
sql = self._query_builder.select(table, columns, where, limit=1)
params = tuple(where.values())
conn = self._pool.acquire()
try:
result = self._executor.execute(conn, sql, params)
if result.success and result.data:
return QueryResult(success=True, data=[result.data[0]])
return QueryResult(success=True, data=[])
finally:
self._pool.release(conn)
def insert(self, table: str, data: Dict[str, Any]) -> QueryResult:
sql = self._query_builder.insert(table, data)
params = tuple(data.values())
conn = self._pool.acquire()
try:
result = self._executor.execute(conn, sql, params)
self._invalidate_cache(table)
return result
finally:
self._pool.release(conn)
def update(
self,
table: str,
data: Dict[str, Any],
where: Dict[str, Any]
) -> QueryResult:
sql = self._query_builder.update(table, data, where)
params = tuple(data.values()) + tuple(where.values())
conn = self._pool.acquire()
try:
result = self._executor.execute(conn, sql, params)
self._invalidate_cache(table)
return result
finally:
self._pool.release(conn)
def delete(self, table: str, where: Dict[str, Any]) -> QueryResult:
sql = self._query_builder.delete(table, where)
params = tuple(where.values())
conn = self._pool.acquire()
try:
result = self._executor.execute(conn, sql, params)
self._invalidate_cache(table)
return result
finally:
self._pool.release(conn)
@contextmanager
def transaction(self):
"""事务上下文管理器"""
tx = TransactionManager(self._executor, self._pool)
try:
tx.begin()
yield tx
tx.commit()
except Exception as e:
tx.rollback()
raise
def _invalidate_cache(self, table: str) -> None:
keys_to_delete = [k for k in self._cache._cache if f":{table}:" in k]
for key in keys_to_delete:
self._cache.delete(key)
@property
def status(self) -> ConnectionStatus:
return self._pool.status
if __name__ == "__main__":
config = ConnectionConfig(
host="localhost",
port=3306,
database="myapp",
username="root",
password="password"
)
db = DatabaseFacade(config)
db.connect()
print("=== 查询所有用户 ===")
result = db.find_all("users", ["id", "name", "email"])
print(f"结果: {result.data}")
print("\n=== 查询单个用户 ===")
result = db.find_one("users", {"id": 1})
print(f"结果: {result.data}")
print("\n=== 插入用户 ===")
result = db.insert("users", {"name": "张三", "email": "zhangsan@example.com"})
print(f"成功: {result.success}")
print("\n=== 事务操作 ===")
with db.transaction() as tx:
tx.execute("INSERT INTO users (name) VALUES (%s)", ("李四",))
tx.execute("UPDATE users SET name = %s WHERE id = %s", ("王五", 1))
db.disconnect()10.4.3 智能家居控制系统
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Callable
from enum import Enum
from datetime import datetime, time
import json
class DeviceState(Enum):
ON = "on"
OFF = "off"
STANDBY = "standby"
class DeviceType(Enum):
LIGHT = "light"
AIR_CONDITIONER = "air_conditioner"
TV = "tv"
CURTAIN = "curtain"
SPEAKER = "speaker"
CAMERA = "camera"
@dataclass
class DeviceStatus:
device_id: str
device_name: str
device_type: DeviceType
state: DeviceState
properties: Dict[str, Any] = field(default_factory=dict)
last_updated: datetime = field(default_factory=datetime.now)
class Light:
"""灯光设备"""
def __init__(self, device_id: str, name: str):
self.device_id = device_id
self.name = name
self.state = DeviceState.OFF
self.brightness = 100
self.color = "#FFFFFF"
def turn_on(self) -> None:
self.state = DeviceState.ON
print(f"💡 {self.name} 已开启")
def turn_off(self) -> None:
self.state = DeviceState.OFF
print(f"💡 {self.name} 已关闭")
def set_brightness(self, level: int) -> None:
self.brightness = max(0, min(100, level))
print(f"💡 {self.name} 亮度: {self.brightness}%")
def set_color(self, color: str) -> None:
self.color = color
print(f"💡 {self.name} 颜色: {color}")
def get_status(self) -> DeviceStatus:
return DeviceStatus(
device_id=self.device_id,
device_name=self.name,
device_type=DeviceType.LIGHT,
state=self.state,
properties={
"brightness": self.brightness,
"color": self.color
}
)
class AirConditioner:
"""空调设备"""
def __init__(self, device_id: str, name: str):
self.device_id = device_id
self.name = name
self.state = DeviceState.OFF
self.temperature = 24.0
self.mode = "cool"
self.fan_speed = "auto"
def turn_on(self) -> None:
self.state = DeviceState.ON
print(f"❄️ {self.name} 已开启")
def turn_off(self) -> None:
self.state = DeviceState.OFF
print(f"❄️ {self.name} 已关闭")
def set_temperature(self, temp: float) -> None:
self.temperature = max(16, min(30, temp))
print(f"❄️ {self.name} 温度: {self.temperature}°C")
def set_mode(self, mode: str) -> None:
self.mode = mode
print(f"❄️ {self.name} 模式: {mode}")
def get_status(self) -> DeviceStatus:
return DeviceStatus(
device_id=self.device_id,
device_name=self.name,
device_type=DeviceType.AIR_CONDITIONER,
state=self.state,
properties={
"temperature": self.temperature,
"mode": self.mode,
"fan_speed": self.fan_speed
}
)
class TV:
"""电视设备"""
def __init__(self, device_id: str, name: str):
self.device_id = device_id
self.name = name
self.state = DeviceState.OFF
self.channel = 1
self.volume = 30
self.source = "hdmi1"
def turn_on(self) -> None:
self.state = DeviceState.ON
print(f"📺 {self.name} 已开启")
def turn_off(self) -> None:
self.state = DeviceState.OFF
print(f"📺 {self.name} 已关闭")
def set_channel(self, channel: int) -> None:
self.channel = channel
print(f"📺 {self.name} 频道: {channel}")
def set_volume(self, volume: int) -> None:
self.volume = max(0, min(100, volume))
print(f"📺 {self.name} 音量: {self.volume}")
def get_status(self) -> DeviceStatus:
return DeviceStatus(
device_id=self.device_id,
device_name=self.name,
device_type=DeviceType.TV,
state=self.state,
properties={
"channel": self.channel,
"volume": self.volume,
"source": self.source
}
)
class Curtain:
"""窗帘设备"""
def __init__(self, device_id: str, name: str):
self.device_id = device_id
self.name = name
self.state = DeviceState.OFF
self.position = 0
def open(self) -> None:
self.state = DeviceState.ON
self.position = 100
print(f"🪟 {self.name} 已打开")
def close(self) -> None:
self.state = DeviceState.OFF
self.position = 0
print(f"🪟 {self.name} 已关闭")
def set_position(self, position: int) -> None:
self.position = max(0, min(100, position))
print(f"🪟 {self.name} 位置: {self.position}%")
def get_status(self) -> DeviceStatus:
return DeviceStatus(
device_id=self.device_id,
device_name=self.name,
device_type=DeviceType.CURTAIN,
state=self.state,
properties={"position": self.position}
)
class Speaker:
"""音响设备"""
def __init__(self, device_id: str, name: str):
self.device_id = device_id
self.name = name
self.state = DeviceState.OFF
self.volume = 50
self.playing = False
def turn_on(self) -> None:
self.state = DeviceState.ON
print(f"🔊 {self.name} 已开启")
def turn_off(self) -> None:
self.state = DeviceState.OFF
self.playing = False
print(f"🔊 {self.name} 已关闭")
def play(self) -> None:
self.playing = True
print(f"🔊 {self.name} 开始播放")
def pause(self) -> None:
self.playing = False
print(f"🔊 {self.name} 已暂停")
def set_volume(self, volume: int) -> None:
self.volume = max(0, min(100, volume))
print(f"🔊 {self.name} 音量: {self.volume}")
def get_status(self) -> DeviceStatus:
return DeviceStatus(
device_id=self.device_id,
device_name=self.name,
device_type=DeviceType.SPEAKER,
state=self.state,
properties={
"volume": self.volume,
"playing": self.playing
}
)
@dataclass
class Scene:
"""场景配置"""
name: str
description: str
actions: List[Callable] = field(default_factory=list)
class SmartHomeFacade:
"""智能家居外观 - 统一控制接口"""
def __init__(self):
self._devices: Dict[str, Any] = {}
self._scenes: Dict[str, Scene] = {}
self._schedules: List[Dict[str, Any]] = []
self._initialize_devices()
self._initialize_scenes()
def _initialize_devices(self) -> None:
self._devices["living_light"] = Light("light_001", "客厅灯")
self._devices["bedroom_light"] = Light("light_002", "卧室灯")
self._devices["study_light"] = Light("light_003", "书房灯")
self._devices["living_ac"] = AirConditioner("ac_001", "客厅空调")
self._devices["bedroom_ac"] = AirConditioner("ac_002", "卧室空调")
self._devices["living_tv"] = TV("tv_001", "客厅电视")
self._devices["living_curtain"] = Curtain("curtain_001", "客厅窗帘")
self._devices["bedroom_curtain"] = Curtain("curtain_002", "卧室窗帘")
self._devices["living_speaker"] = Speaker("speaker_001", "客厅音响")
def _initialize_scenes(self) -> None:
self._scenes["morning"] = Scene(
name="晨间模式",
description="起床模式，打开窗帘，调节温度"
)
self._scenes["movie"] = Scene(
name="观影模式",
description="关闭窗帘，调暗灯光，打开电视"
)
self._scenes["sleep"] = Scene(
name="睡眠模式",
description="关闭所有设备，调节空调温度"
)
self._scenes["leave"] = Scene(
name="离家模式",
description="关闭所有设备"
)
self._scenes["reading"] = Scene(
name="阅读模式",
description="调节灯光，关闭电视"
)
def morning_mode(self) -> None:
"""晨间模式"""
print("\n🌅 晨间模式启动")
self._devices["living_curtain"].open()
self._devices["bedroom_curtain"].open()
self._devices["living_light"].turn_off()
self._devices["bedroom_light"].turn_off()
self._devices["living_ac"].turn_on()
self._devices["living_ac"].set_temperature(24)
self._devices["living_ac"].set_mode("cool")
def movie_mode(self) -> None:
"""观影模式"""
print("\n🎬 观影模式启动")
self._devices["living_curtain"].close()
self._devices["living_light"].turn_on()
self._devices["living_light"].set_brightness(10)
self._devices["living_tv"].turn_on()
self._devices["living_ac"].turn_on()
self._devices["living_ac"].set_temperature(22)
self._devices["living_speaker"].turn_on()
self._devices["living_speaker"].set_volume(40)
def sleep_mode(self) -> None:
"""睡眠模式"""
print("\n🌙 睡眠模式启动")
self._devices["living_light"].turn_off()
self._devices["bedroom_light"].turn_off()
self._devices["study_light"].turn_off()
self._devices["living_tv"].turn_off()
self._devices["living_speaker"].turn_off()
self._devices["living_curtain"].close()
self._devices["bedroom_curtain"].close()
self._devices["bedroom_ac"].turn_on()
self._devices["bedroom_ac"].set_temperature(26)
self._devices["bedroom_ac"].set_mode("sleep")
def leave_home_mode(self) -> None:
"""离家模式"""
print("\n🚪 离家模式启动")
for device in self._devices.values():
if hasattr(device, 'turn_off'):
device.turn_off()
def reading_mode(self) -> None:
"""阅读模式"""
print("\n📖 阅读模式启动")
self._devices["living_tv"].turn_off()
self._devices["living_speaker"].turn_off()
self._devices["study_light"].turn_on()
self._devices["study_light"].set_brightness(80)
self._devices["study_light"].set_color("#FFF5E6")
self._devices["living_curtain"].close()
def all_lights_on(self) -> None:
"""所有灯开启"""
print("\n💡 开启所有灯光")
for key, device in self._devices.items():
if isinstance(device, Light):
device.turn_on()
def all_lights_off(self) -> None:
"""所有灯关闭"""
print("\n💡 关闭所有灯光")
for key, device in self._devices.items():
if isinstance(device, Light):
device.turn_off()
def set_temperature(self, temp: float) -> None:
"""设置所有空调温度"""
print(f"\n🌡️ 设置温度: {temp}°C")
for device in self._devices.values():
if isinstance(device, AirConditioner):
device.set_temperature(temp)
def get_device_status(self, device_id: str) -> Optional[DeviceStatus]:
"""获取设备状态"""
device = self._devices.get(device_id)
if device and hasattr(device, 'get_status'):
return device.get_status()
return None
def get_all_status(self) -> List[DeviceStatus]:
"""获取所有设备状态"""
statuses = []
for device in self._devices.values():
if hasattr(device, 'get_status'):
statuses.append(device.get_status())
return statuses
def get_scene_info(self, scene_name: str) -> Optional[Scene]:
"""获取场景信息"""
return self._scenes.get(scene_name)
def list_scenes(self) -> List[str]:
"""列出所有场景"""
return list(self._scenes.keys())
def execute_scene(self, scene_name: str) -> bool:
"""执行场景"""
scene_methods = {
"morning": self.morning_mode,
"movie": self.movie_mode,
"sleep": self.sleep_mode,
"leave": self.leave_home_mode,
"reading": self.reading_mode,
}
method = scene_methods.get(scene_name)
if method:
method()
return True
return False
def create_custom_scene(
self,
name: str,
actions: List[Dict[str, Any]]
) -> None:
"""创建自定义场景"""
print(f"创建自定义场景: {name}")
self._scenes[name] = Scene(
name=name,
description="自定义场景",
actions=[]
)
def export_status(self) -> str:
"""导出状态为JSON"""
statuses = self.get_all_status()
data = {
"timestamp": datetime.now().isoformat(),
"devices": [
{
"device_id": s.device_id,
"device_name": s.device_name,
"device_type": s.device_type.value,
"state": s.state.value,
"properties": s.properties
}
for s in statuses
]
}
return json.dumps(data, ensure_ascii=False, indent=2)
if __name__ == "__main__":
home = SmartHomeFacade()
home.morning_mode()
home.movie_mode()
home.sleep_mode()
home.leave_home_mode()
print("\n=== 设备状态 ===")
for status in home.get_all_status():
print(f"{status.device_name}: {status.state.value}")
print(f"\n=== 可用场景 ===")
for scene_name in home.list_scenes():
scene = home.get_scene_info(scene_name)
print(f"- {scene.name}: {scene.description}")
print(f"\n=== 导出状态 ===")
print(home.export_status())10.5 企业级应用示例
10.5.1 微服务API网关外观
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Callable
from enum import Enum
from datetime import datetime
import json
import hashlib
class HTTPMethod(Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
DELETE = "DELETE"
PATCH = "PATCH"
@dataclass
class APIRequest:
path: str
method: HTTPMethod
headers: Dict[str, str] = field(default_factory=dict)
params: Dict[str, str] = field(default_factory=dict)
body: Any = None
user_id: Optional[str] = None
@dataclass
class APIResponse:
status_code: int
body: Any
headers: Dict[str, str] = field(default_factory=dict)
execution_time: float = 0.0
class UserService:
"""用户服务"""
def get_user(self, user_id: str) -> Dict[str, Any]:
return {
"id": user_id,
"name": "张三",
"email": "zhangsan@example.com",
"role": "admin"
}
def create_user(self, data: Dict[str, Any]) -> Dict[str, Any]:
return {"id": "user_001", **data}
def update_user(self, user_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
return {"id": user_id, **data}
def delete_user(self, user_id: str) -> bool:
return True
class ProductService:
"""产品服务"""
def get_products(self, params: Dict[str, str]) -> List[Dict[str, Any]]:
return [
{"id": "prod_001", "name": "Python教程", "price": 99.0},
{"id": "prod_002", "name": "设计模式书籍", "price": 128.0},
]
def get_product(self, product_id: str) -> Dict[str, Any]:
return {"id": product_id, "name": "示例产品", "price": 99.0}
def create_product(self, data: Dict[str, Any]) -> Dict[str, Any]:
return {"id": "prod_new", **data}
class OrderService:
"""订单服务"""
def get_orders(self, user_id: str) -> List[Dict[str, Any]]:
return [
{"id": "order_001", "user_id": user_id, "total": 198.0},
{"id": "order_002", "user_id": user_id, "total": 256.0},
]
def create_order(self, data: Dict[str, Any]) -> Dict[str, Any]:
return {"id": "order_new", **data}
def get_order(self, order_id: str) -> Dict[str, Any]:
return {"id": order_id, "status": "pending"}
class PaymentService:
"""支付服务"""
def process_payment(self, order_id: str, amount: float) -> Dict[str, Any]:
return {
"transaction_id": "txn_001",
"order_id": order_id,
"amount": amount,
"status": "completed"
}
def get_payment(self, transaction_id: str) -> Dict[str, Any]:
return {"transaction_id": transaction_id, "status": "completed"}
class NotificationService:
"""通知服务"""
def send_notification(
self,
user_id: str,
message: str
) -> Dict[str, Any]:
return {
"notification_id": "notif_001",
"user_id": user_id,
"message": message,
"sent_at": datetime.now().isoformat()
}
class CacheService:
"""缓存服务"""
def __init__(self):
self._cache: Dict[str, Any] = {}
def get(self, key: str) -> Optional[Any]:
return self._cache.get(key)
def set(self, key: str, value: Any, ttl: int = 300) -> None:
self._cache[key] = value
def delete(self, key: str) -> None:
self._cache.pop(key, None)
class AuthService:
"""认证服务"""
def __init__(self):
self._tokens: Dict[str, str] = {}
def validate_token(self, token: str) -> Optional[str]:
if token.startswith("valid_"):
return token.replace("valid_", "user_")
return None
def generate_token(self, user_id: str) -> str:
token = f"valid_{user_id}"
self._tokens[token] = user_id
return token
class RateLimiter:
"""限流器"""
def __init__(self, max_requests: int = 100, window: int = 60):
self._max_requests = max_requests
self._window = window
self._requests: Dict[str, List[float]] = {}
def is_allowed(self, client_id: str) -> bool:
import time
current = time.time()
if client_id not in self._requests:
self._requests[client_id] = []
self._requests[client_id] = [
t for t in self._requests[client_id]
if current - t < self._window
]
if len(self._requests[client_id]) >= self._max_requests:
return False
self._requests[client_id].append(current)
return True
class APIGatewayFacade:
"""API网关外观 - 统一微服务访问"""
def __init__(self):
self._user_service = UserService()
self._product_service = ProductService()
self._order_service = OrderService()
self._payment_service = PaymentService()
self._notification_service = NotificationService()
self._cache = CacheService()
self._auth = AuthService()
self._rate_limiter = RateLimiter()
def handle_request(self, request: APIRequest) -> APIResponse:
"""处理API请求"""
import time
start_time = time.time()
try:
if not self._rate_limiter.is_allowed(request.user_id or "anonymous"):
return APIResponse(
status_code=429,
body={"error": "请求过于频繁"}
)
if not request.path.startswith("/auth"):
user_id = self._authenticate(request)
if not user_id:
return APIResponse(
status_code=401,
body={"error": "未授权"}
)
request.user_id = user_id
response_body = self._route_request(request)
return APIResponse(
status_code=200,
body=response_body,
execution_time=time.time() - start_time
)
except Exception as e:
return APIResponse(
status_code=500,
body={"error": str(e)},
execution_time=time.time() - start_time
)
def _authenticate(self, request: APIRequest) -> Optional[str]:
token = request.headers.get("Authorization", "")
if token.startswith("Bearer "):
token = token[7:]
return self._auth.validate_token(token)
def _route_request(self, request: APIRequest) -> Any:
path_parts = request.path.strip("/").split("/")
resource = path_parts[0] if path_parts else ""
if resource == "users":
return self._handle_users(request, path_parts)
elif resource == "products":
return self._handle_products(request, path_parts)
elif resource == "orders":
return self._handle_orders(request, path_parts)
elif resource == "auth":
return self._handle_auth(request)
else:
raise ValueError(f"未知资源: {resource}")
def _handle_users(self, request: APIRequest, path_parts: List[str]) -> Any:
if len(path_parts) > 1:
user_id = path_parts[1]
if request.method == HTTPMethod.GET:
return self._user_service.get_user(user_id)
elif request.method == HTTPMethod.PUT:
return self._user_service.update_user(user_id, request.body)
elif request.method == HTTPMethod.DELETE:
return {"success": self._user_service.delete_user(user_id)}
else:
if request.method == HTTPMethod.POST:
return self._user_service.create_user(request.body)
raise ValueError("无效的用户请求")
def _handle_products(self, request: APIRequest, path_parts: List[str]) -> Any:
if len(path_parts) > 1:
product_id = path_parts[1]
return self._product_service.get_product(product_id)
else:
return self._product_service.get_products(request.params)
def _handle_orders(self, request: APIRequest, path_parts: List[str]) -> Any:
if len(path_parts) > 1:
order_id = path_parts[1]
return self._order_service.get_order(order_id)
else:
if request.method == HTTPMethod.GET:
return self._order_service.get_orders(request.user_id)
elif request.method == HTTPMethod.POST:
order = self._order_service.create_order(request.body)
payment = self._payment_service.process_payment(
order["id"],
request.body.get("amount", 0)
)
self._notification_service.send_notification(
request.user_id,
f"订单 {order['id']} 已创建"
)
return {"order": order, "payment": payment}
raise ValueError("无效的订单请求")
def _handle_auth(self, request: APIRequest) -> Any:
if request.method == HTTPMethod.POST:
user_id = request.body.get("user_id", "anonymous")
token = self._auth.generate_token(user_id)
return {"token": token}
raise ValueError("无效的认证请求")
def get_user_with_orders(self, user_id: str) -> Dict[str, Any]:
"""聚合查询 - 用户及其订单"""
user = self._user_service.get_user(user_id)
orders = self._order_service.get_orders(user_id)
return {"user": user, "orders": orders}
def create_order_with_payment(
self,
user_id: str,
items: List[Dict],
amount: float
) -> Dict[str, Any]:
"""创建订单并处理支付"""
order = self._order_service.create_order({
"user_id": user_id,
"items": items,
"amount": amount
})
payment = self._payment_service.process_payment(order["id"], amount)
self._notification_service.send_notification(
user_id,
f"订单 {order['id']} 创建成功，支付金额: ¥{amount}"
)
return {"order": order, "payment": payment}
if __name__ == "__main__":
gateway = APIGatewayFacade()
print("=== 获取用户 ===")
request = APIRequest(
path="/users/user_001",
method=HTTPMethod.GET,
headers={"Authorization": "Bearer valid_user_001"}
)
response = gateway.handle_request(request)
print(f"状态码: {response.status_code}")
print(f"响应: {response.body}")
print("\n=== 获取产品列表 ===")
request = APIRequest(
path="/products",
method=HTTPMethod.GET,
headers={"Authorization": "Bearer valid_user_001"},
params={"category": "books"}
)
response = gateway.handle_request(request)
print(f"产品列表: {response.body}")
print("\n=== 创建订单 ===")
request = APIRequest(
path="/orders",
method=HTTPMethod.POST,
headers={"Authorization": "Bearer valid_user_001"},
body={
"items": [{"product_id": "prod_001", "quantity": 2}],
"amount": 198.0
}
)
response = gateway.handle_request(request)
print(f"订单结果: {response.body}")
print("\n=== 聚合查询 ===")
result = gateway.get_user_with_orders("user_001")
print(f"用户及订单: {result}")10.6 模式变体与扩展
10.6.1 多层外观
python
from typing import Any, Dict, Optional
class SubsystemA:
def operation(self) -> str:
return "SubsystemA"
class SubsystemB:
def operation(self) -> str:
return "SubsystemB"
class LowLevelFacade:
"""底层外观"""
def __init__(self):
self._subsystem_a = SubsystemA()
self._subsystem_b = SubsystemB()
def low_operation(self) -> str:
return f"Low: {self._subsystem_a.operation()} + {self._subsystem_b.operation()}"
class SubsystemC:
def operation(self) -> str:
return "SubsystemC"
class HighLevelFacade:
"""高层外观 - 组合底层外观"""
def __init__(self):
self._low_facade = LowLevelFacade()
self._subsystem_c = SubsystemC()
def high_operation(self) -> str:
low_result = self._low_facade.low_operation()
return f"High: {low_result} + {self._subsystem_c.operation()}"10.6.2 外观工厂
python
from typing import Any, Dict, Type, TypeVar
T = TypeVar('T')
class FacadeFactory:
"""外观工厂"""
_instances: Dict[str, Any] = {}
@classmethod
def get_facade(cls, facade_type: Type[T], name: str = "default") -> T:
key = f"{facade_type.__name__}:{name}"
if key not in cls._instances:
cls._instances[key] = facade_type()
return cls._instances[key]
@classmethod
def clear_all(cls) -> None:
cls._instances.clear()10.6.3 动态外观
python
from typing import Any, Dict, Callable
class DynamicFacade:
"""动态外观 - 运行时注册操作"""
def __init__(self):
self._operations: Dict[str, Callable] = {}
self._subsystems: Dict[str, Any] = {}
def register_subsystem(self, name: str, subsystem: Any) -> None:
self._subsystems[name] = subsystem
def register_operation(self, name: str, handler: Callable) -> None:
self._operations[name] = handler
def __getattr__(self, name: str) -> Any:
if name in self._operations:
return self._operations[name]
raise AttributeError(f"未知操作: {name}")10.7 反模式与最佳实践
10.7.1 常见反模式
反模式1：上帝外观
python
class GodFacade:
"""上帝外观 - 承担过多职责"""
def __init__(self):
self._db = Database()
self._cache = Cache()
self._logger = Logger()
self._email = EmailService()
self._sms = SMSService()
self._payment = PaymentService()
self._shipping = ShippingService()
self._inventory = InventoryService()
self._loyalty = LoyaltyService()
self._analytics = AnalyticsService()
def do_everything(self) -> None:
pass解决方案：拆分为多个专注的外观类。
反模式2：外观膨胀
python
class BloatedFacade:
"""外观膨胀 - 暴露过多方法"""
def method_1(self): pass
def method_2(self): pass
def method_3(self): pass
def method_50(self): pass解决方案：保持外观接口简洁，仅暴露常用操作。
10.7.2 最佳实践
| 实践 | 描述 | 示例 |
|---|---|---|
| 单一职责 | 每个外观专注一个子系统 | OrderFacade, UserFacade |
| 接口简洁 | 仅暴露必要的操作 | 高层业务方法 |
| 不限制访问 | 允许客户端直接访问子系统 | 子系统保持public |
| 分层设计 | 使用多层外观组织复杂度 | LowLevelFacade, HighLevelFacade |
| 依赖注入 | 通过构造函数注入子系统 | __init__(self, subsystem_a, subsystem_b) |
10.8 模式比较
10.8.1 与相关模式对比
| 模式 | 意图 | 结构 | 关键区别 |
|---|---|---|---|
| 外观 | 简化子系统访问 | 单向委托 | 不改变子系统接口 |
| 适配器 | 接口转换 | 单向包装 | 改变接口以兼容 |
| 装饰器 | 动态添加职责 | 链式包装 | 增强功能 |
| 代理 | 控制对象访问 | 代理访问 | 关注访问控制 |
| 中介者 | 协调对象交互 | 集中协调 | 消除对象间直接引用 |
10.8.2 外观模式在架构中的位置
┌─────────────────────────────────────────────────────────────┐
│ 表现层 │
│ (Web/Mobile/API) │
└───────────────────────────────────┬─────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 外观层 │
│ (Facade Layer) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ OrderFacade │ │ UserFacade │ │ProductFacade│ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└───────────────────────────────────┬─────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 业务层 │
│ (Business Layer) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │OrderService │ │UserService │ │ProductService│ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└───────────────────────────────────┬─────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 数据层 │
│ (Data Layer) │
└─────────────────────────────────────────────────────────────┘10.9 决策指南
10.9.1 适用场景检查清单
- [ ] 需要为复杂子系统提供简单接口
- [ ] 客户端与子系统之间存在大量依赖
- [ ] 需要对子系统进行分层
- [ ] 需要封装第三方库的复杂性
- [ ] 希望遵循迪米特法则
10.9.2 实现选择决策树
需要简化什么？
/ \
整个系统 特定子系统
/ \
多层外观 单一外观
/ \ / \
高层外观 底层外观 可配置外观 简单外观10.9.3 快速参考卡
┌─────────────────────────────────────────────────────────────┐
│ 外观模式快速参考 │
├─────────────────────────────────────────────────────────────┤
│ 定义: 为子系统提供统一高层接口，简化访问 │
├─────────────────────────────────────────────────────────────┤
│ 参与者: │
│ • Facade - 简化接口，委托子系统操作 │
│ • Subsystem - 实现专业功能 │
│ • Client - 通过外观访问子系统 │
├─────────────────────────────────────────────────────────────┤
│ 形式化: Facade → ⋃ᵢ Oᵢ (操作组合) │
├─────────────────────────────────────────────────────────────┤
│ 设计原则: │
│ • 迪米特法则 (LoD) │
│ • 单一职责原则 (SRP) │
│ • 接口隔离原则 (ISP) │
├─────────────────────────────────────────────────────────────┤
│ 典型应用: │
│ • API网关 │
│ • 数据库访问层 │
│ • 第三方库封装 │
│ • 复杂系统简化 │
├─────────────────────────────────────────────────────────────┤
│ 最佳实践: │
│ • 保持接口简洁 │
│ • 不限制子系统直接访问 │
│ • 使用依赖注入 │
│ • 避免上帝外观 │
└─────────────────────────────────────────────────────────────┘10.10 小结
外观模式是简化复杂系统接口的有效方式。其核心价值在于：
- 简化接口：为复杂子系统提供简单易用的入口
- 降低耦合：客户端与子系统解耦，遵循迪米特法则
- 分层设计：支持系统的层次化组织
- 灵活访问：不限制客户端直接访问子系统
在Python中，外观模式常用于封装第三方库、简化API调用、构建分层架构。在微服务架构中，API网关是外观模式的现代演进，为分布式系统提供统一的访问入口。
思考与练习
基础练习：为一个文件操作库（读写、压缩、加密）设计外观模式。
进阶练习：实现一个支持插件扩展的外观模式，允许动态注册新的子系统。
挑战练习：设计一个多层外观系统，支持从底层到高层的渐进式简化。
设计思考：外观模式与API网关有何关系？请分析其演进过程。