一 装饰器模式
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"
python">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 = actionreturn funcreturn 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] = funcprint("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
python">from abc import ABC, abstractmethod
import configparser
from inspect import signatureclass Strategy(ABC):@abstractmethoddef execute(self, appoint_order_source):passclass QianShouChaXuanJunWangStrategy(Strategy):def __init__(self, pre_strategy: Strategy = None):self.pre_strategy = pre_strategydef 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_strategydef 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_strategydef 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_strategydef 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}@staticmethoddef 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 = strategydef set_strategy(self, strategy: Strategy):self._strategy = strategydef 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)
python">from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
import logginglogger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)@dataclass
class OrderInfo:product: strquantity: intclass OrderSource(Enum):ONLINE = 'online'OFFLINE = 'offline'class OrderCreator(ABC):"""基类用于创建订单"""@abstractmethoddef create_order_instance(self, order_info: OrderInfo) -> str:"""子类必须实现此方法以创建具体的订单实例"""passclass 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:"""工厂类用于创建不同的订单创建者"""@staticmethoddef 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}")
三 观察者模式
python">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_statusself.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_statusself.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")