一 装饰器模式
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 = 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')
二 工厂模式
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}")