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

python:Coroutines Pattern

news 2026/6/9 22:05:36

项目结构：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述： Coroutines  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/9 21:34 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : settings.py
"""
系统全局配置
"""# 异步流程延时配置（秒）
DELAY_CONFIG = {"purchase": 2,"material_check": 1.5,"process": 3,"authenticate": 2,"sell": 1
}# 鉴定证书前缀
CERTIFICATE_PREFIX = "NGTC"# 日志配置
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述： Coroutines  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/9 21:34 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : logger.py
import logging
from CoroutinesPattern.config.settings import LOG_FORMAT, LOG_DATE_FORMATdef setup_logger():logging.basicConfig(level=logging.INFO,format=LOG_FORMAT,datefmt=LOG_DATE_FORMAT)return logging.getLogger(__name__)# 全局日志实例
logger = setup_logger()# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:35 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : models.py"""
领域实体：珠宝核心模型
"""
from dataclasses import dataclass
from typing import Optional@dataclass
class Jewelry:"""珠宝实体：唯一业务对象"""jewelry_id: strname: strmaterial: strstatus: str = "待采购"certificate: Optional[str] = None

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:36 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : purchase.py"""
采购服务
"""
import asyncio
from CoroutinesPattern.config.settings import DELAY_CONFIG
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.utils.logger import loggerasync def purchase_material(jewelry: Jewelry) -> bool:"""原料采购：仅负责采购业务":param jewelry: :return: """""jewelry.status = "采购中"logger.info(f"【采购服务】开始采购 {jewelry.name}，等待供应商发货...")await asyncio.sleep(DELAY_CONFIG["purchase"])jewelry.status = "原料已到货"logger.info(f"【采购服务】{jewelry.name} 原料到货完成")return True# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:37 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : check.py"""
原料质检服务
"""
import asyncio
from CoroutinesPattern.config.settings import DELAY_CONFIG
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.utils.logger import loggerasync def check_material(jewelry: Jewelry) -> bool:""":param jewelry::return:"""jewelry.status = "原料质检中"logger.info(f"【质检服务】{jewelry.name} 原料检测中...")await asyncio.sleep(DELAY_CONFIG["material_check"])jewelry.status = "原料质检合格"logger.info(f"【质检服务】{jewelry.name} 检测合格")return True# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:38 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : process.py
"""
设计加工服务
"""
import asyncio
from CoroutinesPattern.config.settings import DELAY_CONFIG
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.utils.logger import loggerasync def design_process(jewelry: Jewelry) -> bool:""":param jewelry::return:"""jewelry.status = "加工中"logger.info(f"【加工服务】{jewelry.name} 工匠制作中...")await asyncio.sleep(DELAY_CONFIG["process"])jewelry.status = "成品已完成"logger.info(f"【加工服务】{jewelry.name} 制作完成")return True# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:39 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : auth.py
"""
权威鉴定服务
"""
import asyncio
from CoroutinesPattern.config.settings import DELAY_CONFIG, CERTIFICATE_PREFIX
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.utils.logger import loggerasync def authenticate_jewelry(jewelry: Jewelry) -> bool:""":param jewelry::return:"""jewelry.status = "权威鉴定中"logger.info(f"【鉴定服务】{jewelry.name} 等待证书...")await asyncio.sleep(DELAY_CONFIG["authenticate"])jewelry.certificate = f"{CERTIFICATE_PREFIX}-{jewelry.jewelry_id}-AUTH"jewelry.status = "已获鉴定证书"logger.info(f"【鉴定服务】{jewelry.name} 证书签发：{jewelry.certificate}")return True# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:40 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : sell.py"""
销售出库服务
"""
import asyncio
from CoroutinesPattern.config.settings import DELAY_CONFIG
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.utils.logger import loggerasync def sell_jewelry(jewelry: Jewelry) -> bool:""":param jewelry::return:"""jewelry.status = "待销售"logger.info(f"【销售服务】{jewelry.name} 等待客户付款...")await asyncio.sleep(DELAY_CONFIG["sell"])jewelry.status = "已销售出库"logger.info(f"【销售服务】{jewelry.name} 交易完成，已出库")return True# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:41 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : pipeline.py"""
流程编排：串联所有服务
"""
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.service.purchase import purchase_material
from CoroutinesPattern.service.check import check_material
from CoroutinesPattern.service.process import design_process
from CoroutinesPattern.service.auth import authenticate_jewelry
from CoroutinesPattern.service.sell import sell_jewelry
from CoroutinesPattern.utils.logger import loggerasync def jewelry_full_pipeline(jewelry: Jewelry):"""全业务流程编排职责：仅负责流程顺序，不实现业务逻辑"""logger.info(f"===== 启动 {jewelry.name} 全业务流程 =====")try:await purchase_material(jewelry)await check_material(jewelry)await design_process(jewelry)await authenticate_jewelry(jewelry)await sell_jewelry(jewelry)logger.info(f"===== {jewelry.name} 流程完成 =====")logger.info(f"最终状态：{jewelry.status} | 证书：{jewelry.certificate}\n")except Exception as e:logger.error(f"{jewelry.name} 流程异常：{str(e)}", exc_info=True)

调用：

# encoding: utf-8 
# 版权所有  2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Coroutines  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/9 21:42 
# User      :  geovindu
# Product   : PyCharm
# Project   : pydesginpattern
# File      : CoroutinesBll.pyimport asyncio
from CoroutinesPattern.domain.models import Jewelry
from CoroutinesPattern.workflow.pipeline import jewelry_full_pipeline
from CoroutinesPattern.utils.logger import loggerclass CoroutinesBll(object):""""""async def demo():""":return:"""logger.info("=== 珠宝企业业务系统启动 ===")# 构建业务实体jewelries = [Jewelry("J001", "18K金钻石项链", "黄金+钻石"),Jewelry("J002", "冰种翡翠手镯", "翡翠"),Jewelry("J003", "铂金戒指", "铂金")]# 并发执行全流程tasks = [jewelry_full_pipeline(j) for j in jewelries]await asyncio.gather(*tasks)logger.info("=== 所有珠宝流程全部执行完成 ===")# asyncio.run(demo())

输出：