python: Handshaking Pattern
项目结构:
珠宝订单管理系统实现握手模式
本文展示了一个珠宝订单管理系统的Python实现,采用了"握手模式"来控制工坊负载。系统包含以下核心组件:
- 配置模块:设置工坊最大负载阈值(5)和日志格式
- 枚举类型:定义珠宝类型和订单状态
- 业务异常:处理工坊过载和订单不存在情况
- 日志服务:统一记录系统操作
- 领域层:订单实体和工坊类实现握手模式核心逻辑
- 应用层:订单服务协调业务流程
- 接口层:提供外部调用的API
系统通过"握手模式"确保工坊负载不超过阈值,当工坊满负荷时自动拒绝新订单,完成订单后释放容量可接受新订单,实现了优雅的负载控制机制。
# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:52 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : settings.py """全局配置:企业级统一管理""" class Settings: """ """ # 工坊最大负载(握手模式阈值) WORKSHOP_MAX_LOAD = 5 # 日志配置 LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s" LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" settings = Settings() # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:53 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : jewelry_enums.py """珠宝枚举:可扩展、类型安全""" from enum import Enum class JewelryType(Enum): """ """ DIAMOND_RING = "钻戒" GOLD_BRACELET = "黄金手镯" JADE_PENDANT = "翡翠吊坠" RUBY_NECKLACE = "红宝石项链" PLATINUM_RING = "铂金戒指" SAPPHIRE_EARRING = "蓝宝石耳环" PEARL_NECKLACE = "珍珠项链" class OrderStatus(Enum): """ """ IN_PROGRESS = "生产中" COMPLETED = "已完成" REJECTED = "已拒绝" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:53 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : business_exceptions.py """业务异常:企业级统一异常体系""" class WorkshopOverloadException(Exception): """ 工坊过载异常(握手模式核心) """ pass class OrderNotFoundException(Exception): """ 订单不存在 """ pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:54 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logger.py """基础设施:日志服务""" import logging from HandshakingPattern.config.settings import settings class AppLogger: """ """ def __init__(self): logging.basicConfig( format=settings.LOG_FORMAT, datefmt=settings.LOG_DATE_FORMAT, level=logging.INFO ) self.logger = logging.getLogger("jewelry_system") def info(self, msg: str): self.logger.info(msg) def error(self, msg: str): self.logger.error(msg) logger = AppLogger() # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:55 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : workshop.py """领域层:珠宝工坊(握手模式实现)""" from dataclasses import dataclass from typing import List from HandshakingPattern.config.settings import settings from HandshakingPattern.enums.jewelry_enums import JewelryType, OrderStatus from HandshakingPattern.exceptions.business_exceptions import WorkshopOverloadException,OrderNotFoundException @dataclass class Order: """ 订单实体 """ order_id: str jewelry_type: JewelryType customer: str status: OrderStatus class JewelryWorkshop: def __init__(self): self.max_load = settings.WORKSHOP_MAX_LOAD self.current_orders: List[Order] = [] def handshake(self) -> bool: """ 握手模式核心方法 负载检查:返回是否可接受新订单 """ return len(self.current_orders) < self.max_load def accept_order(self, order: Order) -> None: """接受订单:必须先握手""" if not self.handshake(): raise WorkshopOverloadException( f"工坊已满负荷 {len(self.current_orders)}/{self.max_load}" ) self.current_orders.append(order) def complete_order(self, order_id: str) -> Order: """完成订单,释放负载""" for order in self.current_orders: if order.order_id == order_id: order.status = OrderStatus.COMPLETED self.current_orders.remove(order) return order raise OrderNotFoundException(f"订单{order_id}不存在") def get_status(self): return { "max_load": self.max_load, "current_load": len(self.current_orders), "orders": [o.__dict__ for o in self.current_orders] } # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:57 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : order_service.py """应用层:订单服务(编排业务,不实现核心逻辑)""" from HandshakingPattern.domain.workshop import JewelryWorkshop, Order from HandshakingPattern.enums.jewelry_enums import OrderStatus from HandshakingPattern.infrastructure.logger import logger class OrderService: """ """ def __init__(self, workshop: JewelryWorkshop): self.workshop = workshop def create_order(self, order_id: str, jewelry_type, customer: str) -> str: """ :param order_id: :param jewelry_type: :param customer: :return: """ try: order = Order( order_id=order_id, jewelry_type=jewelry_type, customer=customer, status=OrderStatus.IN_PROGRESS ) self.workshop.accept_order(order) msg = f"✅ 订单{order_id} 接受成功 | 客户:{customer} | 品类:{jewelry_type.value}" logger.info(msg) return msg except Exception as e: msg = f"❌ 订单{order_id} 被拒绝:{str(e)}" logger.error(msg) return msg def finish_order(self, order_id: str) -> str: """ :param order_id: :return: """ try: order = self.workshop.complete_order(order_id) msg = f"🏆 订单{order_id} 已完成 | 客户:{order.customer}" logger.info(msg) return msg except Exception as e: logger.error(str(e)) return str(e) # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述:Handshaking Pattern 握手模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 20:59 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : workshop_api.py """接口层:对外提供服务入口""" from HandshakingPattern.application.order_service import OrderService from HandshakingPattern.domain.workshop import JewelryWorkshop class WorkshopAPI: """ """ def __init__(self): # 依赖注入 self.workshop = JewelryWorkshop() self.order_service = OrderService(self.workshop) def submit_order(self, order_id, jewelry_type, customer): """ :param order_id: :param jewelry_type: :param customer: :return: """ return self.order_service.create_order(order_id, jewelry_type, customer) def complete_order(self, order_id): """ :param order_id: :return: """ return self.order_service.finish_order(order_id) def get_workshop_status(self): """ :return: """ return self.workshop.get_status()调用:
# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看:言語成了邀功盡責的功臣,還需要行爲每日來值班嗎 # 描述: # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/1 21:01 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : HandshakingBll.py from HandshakingPattern.api.workshop_api import WorkshopAPI from HandshakingPattern.enums.jewelry_enums import JewelryType class HandshakingBll(object): """ """ def demo(self): """ :return: """ print("===== 珠宝企业级系统启动(握手模式)=====\n") api = WorkshopAPI() # 模拟订单 orders = [ ("ORD001", JewelryType.DIAMOND_RING, "张先生"), ("ORD002", JewelryType.GOLD_BRACELET, "李女士"), ("ORD003", JewelryType.JADE_PENDANT, "王小姐"), ("ORD004", JewelryType.RUBY_NECKLACE, "刘先生"), ("ORD005", JewelryType.PLATINUM_RING, "陈女士"), ("ORD006", JewelryType.SAPPHIRE_EARRING, "赵先生"), ("ORD007", JewelryType.PEARL_NECKLACE, "孙女士"), ] print("----- 接收客户订单 -----") for oid, jt, cust in orders: print(api.submit_order(oid, jt, cust)) print("\n----- 完成2个订单 -----") print(api.complete_order("ORD001")) print(api.complete_order("ORD002")) print("\n----- 重新接收被拒订单 -----") print(api.submit_order("ORD006", JewelryType.SAPPHIRE_EARRING, "赵先生")) print(api.submit_order("ORD007", JewelryType.PEARL_NECKLACE, "孙女士")) print("\n----- 最终工坊状态 -----") print(api.get_workshop_status())输出
