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

Python高级编程模式和设计模式

article 2025/3/11 16:59:47

一 装饰器模式

order_sources:
  source1:
    on_agreement: "reduce_receivable"
    on_completion: "reduce_receivable"
    on_rejection: "none"
  source2:
    on_agreement: "none"
    on_completion: "reduce_receivable"
    on_rejection: "none"
  source3:
    on_agreement: "none"
    on_completion: "none"
    on_rejection: "none"

import yaml

"""
不同来源执行不同的操作的时候，动态判断进行相应的操作
"""
# 读取配置文件
def load_config(file_path):
    with open(file_path, 'r') as file:
        config = yaml.safe_load(file)
    return config


"""装饰器：装饰器是一种用于修改或增强函数行为的工具。
在这个例子中，register_handler 是一个装饰器工厂，它返回一个装饰器。
装饰器工厂：装饰器工厂可以根据传入的参数动态生成装饰器。
这里，register_handler 接受一个 action 参数，并返回一个装饰器，
该装饰器将 action 属性添加到被装饰的函数上
"""
def register_handler(action):
    def decorator(func):
        func.action = action
        return func

    return decorator


# 定义处理函数
@register_handler("reduce_receivable")
def reduce_receivable(order_id):
    print(f"reduce_receivable receivable for order {order_id}.")


@register_handler("none")
def no_action(order_id):
    print(f"No action11111111 for order {order_id}.")


# 处理类
class AfterSalesHandler:
    def __init__(self, config_file):
        self.config = load_config(config_file)
        self.handlers = self._register_handlers()

    """
    动态注册：_register_handlers 方法遍历全局命名空间中的所有对象，
    查找带有 action 属性的可调用对象，并将它们注册到 handlers 字典中。
    反射：使用 globals() 获取全局命名空间中的所有对象，
    并通过 hasattr 和 callable 进行检查，这是 Python 中的一种反射机制
    """
    def _register_handlers(self):
        handlers = {}
        for name, func in globals().items():
            if callable(func) and hasattr(func, 'action'):
                handlers[func.action] = func
        print("handlers---------", handlers)
        return handlers
    """
    actions：从配置文件中获取指定订单来源的所有操作映射。
    action：从操作映射中获取指定操作对应的动作。
    handler：从注册的处理函数中获取指定动作对应的处理函数。
    handler(order_id)：调用处理函数，并传递订单 ID 作为参数
    """
    def handle_after_sales(self, order_id, order_source, operation):
        actions = self.config['order_sources'].get(order_source, {})
        action = actions.get(operation, 'none')
        handler = self.handlers.get(action, self.nofind_action)
        print("handler-----------------", handler)
        handler(order_id)

    def nofind_action(self, order_id):
        print(f"No action for order {order_id}.")


# 示例调用
if __name__ == "__main__":
    handler = AfterSalesHandler('../data/OrderAfter.yaml')
    handler.handle_after_sales('12345-1', 'source1', 'on_agreement')
    handler.handle_after_sales('12345-2', 'source1', 'on_completion')
    handler.handle_after_sales('12345-3', 'source1', 'on_rejection')
    handler.handle_after_sales('12345-4', 'source2', 'on_agreement')
    handler.handle_after_sales('12345-5', 'source2', 'on_completion')
    handler.handle_after_sales('12345-6', 'source2', 'on_rejection')
    handler.handle_after_sales('12345-7', 'source3', 'on_agreement')
    handler.handle_after_sales('12345-8', 'source3', 'on_completion')
    handler.handle_after_sales('12345-9', 'source3', 'on_rejection')

二 工厂模式

[DynamicMethod]
strategy=CreateJieSuanDan
SignJunWang_pre_strategy=QianShouChaXuanJunWang
HeYanPassJunWang_pre_strategy=SignJunWang
CreateJieSuanDan_pre_strategy=HeYanPassJunWang

from abc import ABC, abstractmethod
import configparser
from inspect import signature


class Strategy(ABC):
    @abstractmethod
    def execute(self, appoint_order_source):
        pass


class QianShouChaXuanJunWangStrategy(Strategy):
    def __init__(self, pre_strategy: Strategy = None):
        self.pre_strategy = pre_strategy

    def execute(self, appoint_order_source):
        if self.pre_strategy:
            self.pre_strategy.execute(appoint_order_source)
        else:
            print('QianShouChaXuanJunWangStrategy没有前置策略')
        print('QianShouChaXuanJunWangStrategy被调用')


class SignJunWangStrategy(Strategy):
    def __init__(self, pre_strategy: Strategy = None):
        self.pre_strategy = pre_strategy

    def execute(self, appoint_order_source):
        if self.pre_strategy:
            self.pre_strategy.execute(appoint_order_source)
        else:
            print('SignJunWangStrategy灭有前置策略')
        print('SignJunWangStrategy被调用')


class HeYanPassJunWangStrategy(Strategy):
    def __init__(self, pre_strategy: Strategy = None):
        self.pre_strategy = pre_strategy

    def execute(self, appoint_order_source):
        if self.pre_strategy:
            self.pre_strategy.execute(appoint_order_source)
        else:
            print('HeYanPassJunWangStrategy没有前置策略')
        print('HeYanPassJunWangStrategy被调用')


class CreateJieSuanDanStrategy(Strategy):
    def __init__(self, pre_strategy: Strategy = None):
        self.pre_strategy = pre_strategy

    def execute(self, appoint_order_source):
        if self.pre_strategy:
            self.pre_strategy.execute(appoint_order_source)
        else:
            print('CreateJieSuanDanStrategy没有前置策略')
        print('CreateJieSuanDanStrategy被调用')


"""
最终返回的策略对象链是:
CreateJieSuanDanStrategy(
    HeYanPassJunWangStrategy(
        SignJunWangStrategy(
            QianShouChaXuanJunWangStrategy(None)
        )
    )
)
"""


class StrategyFactory:
    _strategies = {
        'QianShouChaXuanJunWang': QianShouChaXuanJunWangStrategy,
        'SignJunWang': SignJunWangStrategy,
        'HeYanPassJunWang': HeYanPassJunWangStrategy,
        'CreateJieSuanDan': CreateJieSuanDanStrategy
    }

    @staticmethod
    def create_strategy(strategy_name: str, config: configparser.ConfigParser) -> Strategy:
        if strategy_name not in StrategyFactory._strategies:
            raise ValueError(f"Unknown strategy: {strategy_name}")
        strategy_class = StrategyFactory._strategies[strategy_name]
        pre_strategy_name = config.get('DynamicMethod', f'{strategy_name}_pre_strategy', fallback=None)
        if pre_strategy_name:
            pre_strategy = StrategyFactory.create_strategy(pre_strategy_name, config)
            return strategy_class(pre_strategy)
        else:
            return strategy_class()


class Context:
    def __init__(self, strategy: Strategy):
        self._strategy = strategy

    def set_strategy(self, strategy: Strategy):
        self._strategy = strategy

    def execute_strategy(self, appoint_order_source):
        self._strategy.execute(appoint_order_source)


def dynamic_method_call(appoint_order_source, config_file='../data/DongtaiConfig.ini'):
    # 读取配置文件
    config = configparser.ConfigParser()
    config.read(config_file)

    # 获取当前需要调用的策略名称
    strategy_name = config.get('DynamicMethod', 'strategy')

    # 创建策略对象
    strategy = StrategyFactory.create_strategy(strategy_name, config)
    # 创建上下文并设置策略
    context = Context(strategy)

    # 执行策略
    context.execute_strategy(appoint_order_source)


if __name__ == "__main__":
    appoint_order_source = "some_value"

    # 动态调用方法
    dynamic_method_call(appoint_order_source)


from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


@dataclass
class OrderInfo:
    product: str
    quantity: int


class OrderSource(Enum):
    ONLINE = 'online'
    OFFLINE = 'offline'


class OrderCreator(ABC):
    """基类用于创建订单"""

    @abstractmethod
    def create_order_instance(self, order_info: OrderInfo) -> str:
        """子类必须实现此方法以创建具体的订单实例"""
        pass


class OnlineOrderCreator(OrderCreator):
    """用于创建线上订单的类"""

    def create_order_instance(self, order_info: OrderInfo) -> str:
        """实现线上订单的创建逻辑"""
        logger.info(f"Creating online order: {order_info}")
        return "OnlineOrder123"


class OfflineOrderCreator(OrderCreator):
    """用于创建线下订单的类"""

    def create_order_instance(self, order_info: OrderInfo) -> str:
        """实现线下订单的创建逻辑"""
        logger.info(f"Creating offline order: {order_info}")
        return "OfflineOrder456"


class OrderFactory:
    """工厂类用于创建不同的订单创建者"""

    @staticmethod
    def get_creator(source: OrderSource) -> OrderCreator:
        """根据来源获取相应的订单创建者实例"""
        if source == OrderSource.ONLINE:
            return OnlineOrderCreator()
        elif source == OrderSource.OFFLINE:
            return OfflineOrderCreator()
        else:
            raise ValueError(f"Invalid order source: '{source.value}'")


def create_order_instance(order_info: OrderInfo, source: OrderSource) -> str:
    """根据指定的来源创建订单实例"""
    creator = OrderFactory.get_creator(source)
    return creator.create_order_instance(order_info)


# 使用示例
order_info = OrderInfo(product="Widget", quantity=10)
source = OrderSource.ONLINE
order_id = create_order_instance(order_info, source)
print(f"Order ID: {order_id}")

三 观察者模式

class OrderSubject:
    def __init__(self):
        self._observers = []
        self._orders = {}

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self, order_id, status):
        for observer in self._observers:
            observer.update(order_id, status)

    def create_order(self, order_id, initial_status):
        self._orders[order_id] = initial_status
        self.notify(order_id, initial_status)

    def change_order_status(self, order_id, new_status):
        if order_id in self._orders:
            self._orders[order_id] = new_status
            self.notify(order_id, new_status)
        else:
            print(f"Order {order_id} does not exist.")


class LogisticsObserver:
    def update(self, order_id, status):
        print(f"Logistics: Order {order_id} status changed to {status}")
        if status == "Shipped":
            self.ship_order(order_id)

    def ship_order(self, order_id):
        print(f"Logistics: Shipping order {order_id}")


class FinanceObserver:
    def update(self, order_id, status):
        print(f"Finance: Order {order_id} status changed to {status}")
        if status == "Paid":
            self.process_payment(order_id)

    def process_payment(self, order_id):
        print(f"Finance: Processing payment for order {order_id}")


class CustomerNotificationObserver:
    def update(self, order_id, status):
        print(f"Customer: Order {order_id} status changed to {status}")
        if status == "Delivered":
            self.notify_customer(order_id)

    def notify_customer(self, order_id):
        print(f"Customer: Notifying customer about delivery of order {order_id}")


# 创建订单主体
order_subject = OrderSubject()

# 创建观察者
logistics_observer = LogisticsObserver()
finance_observer = FinanceObserver()
customer_notification_observer = CustomerNotificationObserver()

# 注册观察者
order_subject.attach(logistics_observer)
order_subject.attach(finance_observer)
order_subject.attach(customer_notification_observer)

# 创建订单
order_subject.create_order("12345", "Created")

# 改变订单状态
order_subject.change_order_status("12345", "Paid")
order_subject.change_order_status("12345", "Shipped")
order_subject.change_order_status("12345", "Delivered")

http://www.kler.cn/a/400424.html

相关文章：

  • CTF攻防世界小白刷题自学笔记14
  • Flink算子
  • Vue 3 中 ref 属性详解：操作 DOM 元素的利器
  • Python的3D可视化库 - vedo (1)简介和模块功能概览
  • ThinkPHP6的ORM模型
  • hive-内部表外部表-详细介绍
  • Java 网络编程：Socket 与网络通信
  • Jtti：服务器总是自动重启怎么办?
  • 如何保存python文件
  • 最新6.7分非肿瘤纯生信，使用机器学习筛选慢阻肺中的关键基因。机器学习在非肿瘤生信文章中正火，可重复！
  • Python自动化DevOps任务入门
  • stm32学习笔记----51单片机和stm32单片机的区别
  • w043基于springboot的“衣依”服装销售平台的设计与实现
  • postgresql（功能最强大的开源数据库）继承特性和分区实现
  • STM32 ADC --- DMA乒乓缓存
  • Spark：大数据处理的强大引擎
  • Elasticsearch的查询语法——DSL 查询
  • 网页作业9
  • esp32c3开发板通过micropython的mqtt库连MQTT物联网消息服务器
  • @JsonSerialize修复前端精度问题