项目结构# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:09 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : exceptions.py class BusinessException(Exception): 业务基础异常 pass class ServiceTimeoutException(BusinessException): 服务超时异常截止日期模式触发 pass class ServiceInvokeException(BusinessException): 服务调用失败异常 pass # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:10 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logger.py import logging import sys def get_logger(name: str) - logging.Logger: 企业日志 :param name: :return: logger logging.getLogger(name) logger.setLevel(logging.INFO) formatter logging.Formatter( %(asctime)s | %(levelname)s | %(name)s | %(message)s ) console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) logger.propagate False return logger # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:10 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : deadline.py import threading from typing import Callable, Any, Generic, TypeVar from DeadlinePattern.common.exceptions import ServiceTimeoutException, ServiceInvokeException from DeadlinePattern.common.logger import get_logger logger get_logger(DeadlinePattern) T TypeVar(T) class DeadlineExecutor(Generic[T]): 截止日期模式企业级核心 企业级截止日期模式执行器 职责统一处理超时任务超时终止等待保证系统稳定性 def __init__(self, timeout_seconds: float): self.timeout timeout_seconds self.result: T | None None self.error: Exception | None None def _run(self, func: Callable[[], T]): try: self.result func() except Exception as e: self.error e def execute(self, func: Callable[[], T]) - T: thread threading.Thread(targetself._run, args(func,)) thread.start() thread.join(self.timeout) if thread.is_alive(): logger.error(f任务超时已终止等待超时{self.timeout}s) raise ServiceTimeoutException(超时响应概率极低主动终止等待) if self.error: raise ServiceInvokeException(f服务执行异常{str(self.error)}) logger.info(任务执行完成) return self.result # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:08 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : config.py # 全局超时配置截止日期模式超时时间 GEM_IDENTIFY_TIMEOUT_SECONDS 3 # 业务配置 ORDER_ID_PREFIX JEWELRY_ORDER # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:13 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : models.py from dataclasses import dataclass from typing import Optional dataclass class GemIdentifyResult: 宝石鉴定结果实体 gem_type: str carat: float color: str clarity: str identify_no: str dataclass class JewelryOrder: 珠宝订单实体 order_id: str identify_result: Optional[GemIdentifyResult] None material_approved: bool False produce_finished: bool False quality_passed: bool False delivered: bool False# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 21:00 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : gem_identify_service.py import random import time from DeadlinePattern.domain.models import GemIdentifyResult from DeadlinePattern.common.logger import get_logger logger get_logger(GemIdentifyService) class GemIdentifyService: 第三方宝石鉴定服务外部不稳定依赖 def identify(self) - GemIdentifyResult: :return: logger.info(开始调用第三方宝石鉴定服务...) # 随机控制响应速度50%快速50%超长耗时触发截止期限 if random.random() 0.5: delay 1 else: delay 10 time.sleep(delay) res GemIdentifyResult( gem_type钻石, carat1.0, colorD, clarityVVS1, identify_nofGEM_{random.randint(100000,999999)} ) logger.info(f鉴定完成{res}) return res class GemIdentifyService: 第三方宝石鉴定服务外部不稳定依赖 def identify(self) - GemIdentifyResult: :return: logger.info(开始调用第三方宝石鉴定服务...) # 模拟不稳定50% 1秒返回50% 10秒返回 delay 1 if random.random() 0.5 else 10 time.sleep(delay) result GemIdentifyResult( gem_type钻石, carat1.0, colorD, clarityVVS1, identify_nofGEM_{random.randint(100000, 999999)} ) logger.info(f鉴定完成{result}) return result # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 21:02 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : material_service.py from DeadlinePattern.common.logger import get_logger logger get_logger(MaterialService) class MaterialService: def approve(self) - bool: logger.info(原料采购审核中...) return True # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 21:03 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : produce_service.py from DeadlinePattern.common.logger import get_logger logger get_logger(ProduceService) class ProduceService: def produce(self) - bool: :return: logger.info(工厂加工生产完成) return True # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 21:04 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : quality_service.py from DeadlinePattern.common.logger import get_logger logger get_logger(QualityService) class QualityService: def check(self) - bool: :return: logger.info(成品质检合格) return True # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 21:05 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : deliver_service.py from DeadlinePattern.common.logger import get_logger logger get_logger(DeliverService) class DeliverService: def deliver(self) - bool: :return: logger.info(订单已交付) return True # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述Deadline 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/6/29 20:08 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : workflow_manager.py from DeadlinePattern.domain.models import JewelryOrder, GemIdentifyResult from DeadlinePattern.service.external.gem_identify_service import GemIdentifyService from DeadlinePattern.service.internal.material_service import MaterialService from DeadlinePattern.service.internal.produce_service import ProduceService from DeadlinePattern.service.internal.quality_service import QualityService from DeadlinePattern.service.internal.deliver_service import DeliverService from DeadlinePattern.common.deadline import DeadlineExecutor from DeadlinePattern.common.exceptions import ServiceTimeoutException from DeadlinePattern.common.logger import get_logger from DeadlinePattern.config.config import GEM_IDENTIFY_TIMEOUT_SECONDS logger get_logger(WorkflowManager) class JewelryWorkflowManager: 流程编排器核心 珠宝全流程管理器 职责统一编排所有服务不实现业务逻辑 def __init__(self): self.identify GemIdentifyService() self.material MaterialService() self.produce ProduceService() self.quality QualityService() self.deliver DeliverService() # def run(self, order: JewelryOrder): :param order: :return: logger.info(f开始执行订单全流程{order.order_id}) try: exec DeadlineExecutor[GemIdentifyResult](GEM_IDENTIFY_TIMEOUT_SECONDS) order.identify_result exec.execute(self.identify.identify) except ServiceTimeoutException: logger.error(鉴定超时启动人工复核兜底流程) order.material_approved self.material.approve() order.produce_finished self.produce.produce() order.quality_passed self.quality.check() order.delivered self.deliver.deliver() logger.info(f订单全流程执行完成{order.order_id}) return order class JewelryWorkflowManager: 流程编排器核心 珠宝全流程管理器 职责统一编排所有服务不实现业务逻辑 def __init__(self): # 依赖注入可扩展、可测试 self.identify_service GemIdentifyService() self.material_service MaterialService() self.produce_service ProduceService() self.quality_service QualityService() self.deliver_service DeliverService() def execute_full_workflow(self, order: JewelryOrder) - JewelryOrder: :param order: :return: logger.info(f开始执行订单全流程{order.order_id}) # 截止日期模式鉴定服务 try: executor DeadlineExecutor[GemIdentifyResult](GEM_IDENTIFY_TIMEOUT_SECONDS) identify_result executor.execute(self.identify_service.identify) order.identify_result identify_result except ServiceTimeoutException: logger.error(鉴定超时启动人工复核兜底流程) # 企业级兜底逻辑不阻塞主流程 # 内部稳定流程 order.material_approved self.material_service.approve() order.produce_finished self.produce_service.produce() order.quality_passed self.quality_service.check() order.delivered self.deliver_service.deliver() logger.info(f订单全流程执行完成{order.order_id}) return order 调用# 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/6/29 21:07 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : DeadlineBll.py import time import random from DeadlinePattern.domain.models import JewelryOrder from DeadlinePattern.core.workflow_manager import JewelryWorkflowManager from DeadlinePattern.common.logger import get_logger from DeadlinePattern.config.config import ORDER_ID_PREFIX logger get_logger(DeadlineBll) class DeadlineBll(object): def demo(self): :return: print( 展示【Deadline Pattern截止期限模式】示例 ) logger.info( 珠宝定制系统启动 ) order JewelryOrder(f{ORDER_ID_PREFIX}_{random.randint(1000, 9999)}) logger.info(f订单创建{order}) # 这里现在一定正常 manager JewelryWorkflowManager() final manager.run(order) logger.info( * 60) logger.info(f最终订单状态{final}) logger.info( 系统运行完成 ) logger.info( 珠宝定制系统启动 ) # 1. 创建订单 order JewelryOrder( order_idf{ORDER_ID_PREFIX}_{random.randint(1000, 9999)} ) logger.info(f订单创建{order}) # 2. 执行全流程 manager JewelryWorkflowManager() final_order manager.execute_full_workflow(order) # 3. 最终结果 logger.info( * 60) logger.info(f最终订单状态{final_order}) logger.info( 系统运行完成 ) time.sleep(1) 输出