Python六边形架构

# Python 六边形架构 (Hexagonal Architecture)# 端口与适配器模式核心业务逻辑与外部世界完全隔离# 驱动适配器 - [输入端口] - 核心领域 - [输出端口] - 被驱适配器from abc import ABC, abstractmethodfrom dataclasses import dataclass# 核心领域dataclassclass Order:id: stramount: floatstatus: str pendingdef approve(self) - None:if self.amount 0:raise ValueError(金额不能为负)self.status approved# 端口 (Port)class OrderRepositoryPort(ABC):abstractmethoddef save(self, order: Order) - None: ...abstractmethoddef find_by_id(self, order_id: str) - Order | None: ...class PaymentServicePort(ABC):abstractmethoddef charge(self, amount: float) - bool: ...class NotifierPort(ABC):abstractmethoddef send(self, order: Order) - None: ...# 核心应用服务class ApproveOrderUseCase:def __init__(self, repo: OrderRepositoryPort,payment: PaymentServicePort, notify: NotifierPort):self._repo repo; self._payment payment; self._notify notifydef execute(self, order_id: str) - Order:order self._repo.find_by_id(order_id)if order is None:raise ValueError(订单不存在)order.approve()if not self._payment.charge(order.amount):raise RuntimeError(支付失败)self._repo.save(order)self._notify.send(order)return order# 被驱适配器 (Driven Adapter)class InMemoryOrderRepo(OrderRepositoryPort):def __init__(self):self._store: dict[str, Order] {}def save(self, o: Order) - None:self._store[o.id] odef find_by_id(self, oid: str) - Order | None:return self._store.get(oid)class MockPaymentService(PaymentServicePort):def charge(self, amount: float) - bool:return Trueclass EmailNotifier(NotifierPort):def send(self, order: Order) - None:print(f[通知] 订单 {order.id} 已确认)# 驱动适配器 (Driving Adapter)class FastAPIAdapter:def __init__(self, use_case: ApproveOrderUseCase):self._use_case use_casedef post_approve(self, order_id: str) - dict:order self._use_case.execute(order_id)return {id: order.id, status: order.status}# 组合根 (Composition Root)def bootstrap() - FastAPIAdapter:repo InMemoryOrderRepo()payment MockPaymentService()notifier EmailNotifier()return FastAPIAdapter(ApproveOrderUseCase(repo, payment, notifier))if __name__ __main__:app bootstrap()repo InMemoryOrderRepo()repo.save(Order(ORD-001, 299.99))result app.post_approve(ORD-001)print(result)