当前位置： 首页 > news >正文

python: Deadline Pattern

news 2026/6/30 1:29:56

项目结构：

# 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(target=self._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="钻石", carat=1.0, color="D", clarity="VVS1", identify_no=f"GEM_{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="钻石", carat=1.0, color="D", clarity="VVS1", identify_no=f"GEM_{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_id=f"{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) '''

输出：

http://www.jsqmd.com/news/1092917/

相关文章：

  • 从零到一：如何用免费开源Verilog工具链打造专业数字电路
  • StockWidget：桌面悬浮的轻量盯盘小工具
  • 关于vidocoding的开发流程
  • 微信小程序云开发实战：从0到1构建“商业清洁预约”双向匹配后端
  • CBDC安全架构：密码学签名与硬件防护核心技术解析
  • 【单片机毕业设计】基于 STM32 的多模式智能路灯控制系统设计， 基于单片机的光照自适应路灯亮度调节系统设计（014001）
  • Python 文件打开模式总结
  • 为什么顶尖AI团队拒绝“通用提示词”？——稀缺首发：金融/医疗/法律三大垂直领域217条经审计Prompt资产包（限时开放下载）
  • 图片进知识库：先让模型生成文字描述再检索
  • StyleGAN 技术脉络：从风格空间到无混叠生成
  • 《科技代替了我工作》值得被认真放进中文歌单
  • Java 多线程：继承 Thread 与实现 Runnable 两种创建方式完整对比
  • 吾爱出品，必属精品，离线可用！
  • 自动定期备份服务器数据
  • python下载M3U8视频脚本
  • 【单片机毕业设计】基于 STM32 的超重声光报警电子秤设计与实现，基于 STM32 的阈值式重量监测报警系统设计（013701）
  • AI截图工具免费下载，基于DeepSeek的OCR截图软件支持Mac和Win
  • 三列布局三大方案对比总结
  • TI TPIC7710EVM评估模块：汽车EPB系统ASIC驱动与电机控制实战解析
  • Burp Suite实战：验证码场景下的自动化渗透测试与绕过技术
  • 专业iOS激活锁绕过工具applera1n：5分钟恢复iPhone 6s-X使用权限
  • 单板测试正常，整机运行异常：6 类系统排查清单
  • 关于我的第十次web作业
  • YOLO轻量化与部署优化- 第75篇：ONNX导出与优化：从PyTorch到ONNX部署
  • 传统潮流款库存一定会亏损，编程潮流款二手转售，改款二次销售收益模型，降低滞销亏损。
  • PilotGo-plugin-llmops API全解析：轻松集成与二次开发
  • ABB工业机器人编程基础（十三）功能程序（FUNC）
  • 合肥第三期《AI产品经理训练营》授课心得
  • 考四级的资料|过四级必备资料书|英语六级备考资料
  • MySQL数据库期末复习②