Python 工厂模式

1. 什么是工厂方法模式（Factory Method）？

工厂方法模式属于创建型设计模式。它的核心意图是定义一个用于创建对象的接口，让子类决定实例化哪一个类。

• 传统方式：客户端直接使用 new 或构造函数（如 obj = EmailSender()）硬编码创建对象。
• 工厂方法：引入一个“工厂”抽象层，由具体工厂子类去决定具体创建哪一个产品对象。

2. 角色划分

• Product（抽象产品）：定义产品的公共接口或基类。
• ConcreteProduct（具体产品）：实现抽象产品接口的具体类。
• Creator（抽象工厂）：声明工厂方法，该方法返回一个 Product 类型的对象。
• ConcreteCreator（具体工厂）：重写工厂方法，返回一个 ConcreteProduct 实例。

---

二、 标准实现：从零构建一个通知系统

假设我们要开发一个通知系统，支持 Email（邮件）、SMS（短信） 和 Push（App推送） 三种通知方式。如果未来要加 Slack 或微信通知，我们需要做到对修改关闭，对扩展开放。

标准实现代码

from abc import ABC, abstractmethod
from typing import Optional, Dict, Type
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class Notifier(ABC):
    """通知器抽象基类"""
    
    @abstractmethod
    def send(self, message: str) -> bool:
        """
        发送消息
        :param message: 消息内容
        :return: 发送成功返回True，失败返回False
        """
        pass


class EmailNotifier(Notifier):
    """邮件通知器"""
    
    def send(self, message: str) -> bool:
        try:
            print(f"[Email系统] 发送邮件内容: {message}")
            # 这里可以添加实际的邮件发送逻辑
            logger.info("邮件发送成功")
            return True
        except Exception as e:
            logger.error(f"邮件发送失败: {str(e)}")
            return False


class SMSNotifier(Notifier):
    """短信通知器"""
    
    def send(self, message: str) -> bool:
        try:
            print(f"[SMS系统] 发送短信内容: {message}")
            # 这里可以添加实际的短信发送逻辑
            logger.info("短信发送成功")
            return True
        except Exception as e:
            logger.error(f"短信发送失败: {str(e)}")
            return False


class PushNotifier(Notifier):
    """APP推送通知器"""
    
    def send(self, message: str) -> bool:
        try:
            print(f"[Push系统] 发送APP推送内容: {message}")
            # 这里可以添加实际的推送发送逻辑
            logger.info("APP推送发送成功")
            return True
        except Exception as e:
            logger.error(f"APP推送发送失败: {str(e)}")
            return False


class NotificationFactory(ABC):
    """通知工厂抽象基类（结合模板方法模式）"""
    
    @abstractmethod
    def create_notifier(self) -> Notifier:
        """创建通知器实例（工厂方法）"""
        pass

    def deliver(self, message: str, recipient: Optional[str] = None) -> bool:
        """
        发送通知的模板方法：定义完整的发送流程
        :param message: 消息内容
        :param recipient: 接收者（可选）
        :return: 发送是否成功
        """
        notifier = self.create_notifier()
        
        # 前置处理：校验消息
        if not self._validate_message(message):
            logger.error("消息校验失败：消息内容为空")
            return False
        
        # 前置处理：格式化消息
        formatted_message = self._format_message(message, recipient)
        print(f"-> 前置处理: 校验消息内容合法性通过，消息已格式化")
        
        # 核心逻辑：发送消息
        try:
            success = notifier.send(formatted_message)
        except Exception as e:
            logger.error(f"发送过程出现异常: {str(e)}")
            success = False
        
        # 后置处理
        if success:
            print("-> 后置处理: 发送成功，记录日志")
            self._log_success(formatted_message, recipient)
        else:
            print("-> 后置处理: 发送失败，记录错误日志")
            self._log_failure(formatted_message, recipient)
            
        return success

    def _validate_message(self, message: str) -> bool:
        """校验消息合法性（可被子类重写）"""
        return bool(message and message.strip())

    def _format_message(self, message: str, recipient: Optional[str] = None) -> str:
        """格式化消息（可被子类重写）"""
        if recipient:
            return f"致 {recipient}: {message}"
        return message

    def _log_success(self, message: str, recipient: Optional[str]) -> None:
        """记录成功日志"""
        logger.info(f"通知发送成功 | 接收者: {recipient or '未指定'} | 内容: {message[:30]}...")

    def _log_failure(self, message: str, recipient: Optional[str]) -> None:
        """记录失败日志"""
        logger.error(f"通知发送失败 | 接收者: {recipient or '未指定'} | 内容: {message[:30]}...")


class EmailNotificationFactory(NotificationFactory):
    """邮件通知工厂"""
    
    def create_notifier(self) -> Notifier:
        return EmailNotifier()

    def _format_message(self, message: str, recipient: Optional[str] = None) -> str:
        """邮件特有的格式化逻辑"""
        header = "【邮件通知】\n"
        if recipient:
            return f"{header}尊敬的{recipient}，您好！\n{message}\n\n此邮件为系统自动发送，请勿回复。"
        return f"{header}{message}\n\n此邮件为系统自动发送，请勿回复。"


class SMSNotificationFactory(NotificationFactory):
    """短信通知工厂"""
    
    def create_notifier(self) -> Notifier:
        return SMSNotifier()

    def _format_message(self, message: str, recipient: Optional[str] = None) -> str:
        """短信特有的格式化逻辑，控制长度"""
        prefix = "【短信提醒】"
        msg = f"{prefix}{message}"
        # 模拟短信长度限制
        return msg[:70] + ("..." if len(msg) > 70 else "")


class PushNotificationFactory(NotificationFactory):
    """APP推送通知工厂"""
    
    def create_notifier(self) -> Notifier:
        return PushNotifier()

    def _format_message(self, message: str, recipient: Optional[str] = None) -> str:
        """推送特有的格式化逻辑"""
        return f"📱 {message}"


# ===== 扩展：简单工厂 + 注册机制（可选增强）=====
class NotificationFactoryRegistry:
    """通知工厂注册中心（简单工厂模式实现）"""
    
    _factories: Dict[str, Type[NotificationFactory]] = {}

    @classmethod
    def register(cls, notification_type: str, factory_class: Type[NotificationFactory]) -> None:
        """注册通知类型"""
        cls._factories[notification_type.lower()] = factory_class
        logger.debug(f"已注册通知类型: {notification_type}")

    @classmethod
    def get_factory(cls, notification_type: str) -> NotificationFactory:
        """根据类型获取工厂实例"""
        factory_class = cls._factories.get(notification_type.lower())
        if not factory_class:
            raise ValueError(f"不支持的通知类型: {notification_type}，支持的类型: {list(cls._factories.keys())}")
        return factory_class()

    @classmethod
    def send_notification(cls, notification_type: str, message: str, recipient: Optional[str] = None) -> bool:
        """便捷方法：直接发送通知"""
        factory = cls.get_factory(notification_type)
        return factory.deliver(message, recipient)


# 注册所有通知类型
NotificationFactoryRegistry.register("email", EmailNotificationFactory)
NotificationFactoryRegistry.register("sms", SMSNotificationFactory)
NotificationFactoryRegistry.register("push", PushNotificationFactory)


if __name__ == "__main__":
    print("=" * 60)
    print("--- 测试邮件通知流程 ---")
    email_factory = EmailNotificationFactory()
    email_factory.deliver("您的验证码是：1234，5分钟内有效，请勿泄露给他人。", recipient="张三")

    print("\n" + "=" * 60)
    print("--- 测试短信通知流程 ---")
    sms_factory = SMSNotificationFactory()
    sms_factory.deliver("您有一条新的快递待取，取件码：A12-3-456，请凭取件码到快递柜领取。", recipient="138‌****‌8888")

    print("\n" + "=" * 60)
    print("--- 测试APP推送通知流程 ---")
    push_factory = PushNotificationFactory()
    push_factory.deliver("您关注的商品已降价，快来查看吧！", recipient="用户_12345")

    print("\n" + "=" * 60)
    print("--- 使用注册中心统一发送（推荐方式）---")
    NotificationFactoryRegistry.send_notification(
        "email", 
        "您的订单已发货，预计明日送达。", 
        recipient="李四"
    )
    
    print("\n" + "=" * 60)
    print("--- 测试异常处理：空消息 ---")
    NotificationFactoryRegistry.send_notification("sms", "", recipient="139‌****‌9999")

    print("\n" + "=" * 60)
    print("--- 测试异常处理：不支持的类型 ---")
    try:
        NotificationFactoryRegistry.send_notification("wechat", "测试消息")
    except ValueError as e:
        print(f"预期内的错误: {e}")

逐行深度解析

1. **Notifier(ABC)**：规范了所有通知产品必须实现 send 方法，这是面向接口编程的基石。
2. NotificationFactory(ABC)：不仅定义了抽象的 create_notifier，还通过 deliver 方法展示了设计模式的精髓——控制反转（IoC）。业务流程由抽象工厂掌控，而具体造什么对象由子类工厂决定。
3. 隔离变化：如果未来增加 WeChatNotifier，我们不需要修改任何原有的业务代码，只需新增一个 WeChatNotifier 类和一个 WeChatNotificationFactory 类即可。

---

三、 企业级痛点：传统的“类爆炸”问题

虽然标准工厂方法模式非常优雅，但在真实企业级大项目中，它有一个致命的温床：类爆炸（Class Explosion）。

• 如果你有 10 种支付方式（支付宝、微信、ApplePay、PayPal……），标准工厂模式需要写 10 个产品类 + 10 个工厂类 = 20 个类。
• 如果业务频繁迭代，每天新增 3 种支付渠道，开发者会淹没在写各种 XxxFactory 子文件的机械劳动中。

企业级解决方案：引入注册表模式（Registry Pattern）与动态反射机制，将“类工厂”简化为“数据驱动的注册中心”。

---

四、 企业级高阶实战：动态注册与配置驱动的支付系统

这是一个融合了工厂模式 + 注册表模式 + 动态映射的企业级生产架构。支持通过字符串配置或前端传参，动态选择策略并实例化。

企业级架构代码

from abc import ABC, abstractmethod
from typing import Dict, Type


# ==================== 1. 抽象产品与具体产品 ====================
class Payment(ABC):

  @abstractmethod
  def pay(self, amount: float):
    pass


class WeChatPayment(Payment):

  def pay(self, amount: float):
    print(f"[企业级] 成功发起微信支付: ¥{amount}")


class AliPayPayment(Payment):

  def pay(self, amount: float):
    print(f"[企业级] 成功发起支付宝支付: ¥{amount}")


class PayPalPayment(Payment):

  def pay(self, amount: float):
    print(f"[企业级] 成功发起PayPal支付: ${amount}")


# ==================== 2. 企业级集中式工厂与注册中心 ====================
class EnterprisePaymentFactory:
  # 内存注册表：用于集中存储“类型名称 -> 产品类”的映射
  _registry: Dict[str, Type[Payment]] = {}

  @classmethod
  def register(cls, payment_type: str):
    """装饰器：用于自动注册具体产品类到工厂中，消灭子类工厂膨胀"""

    def decorator(subclass: Type[Payment]):
      if payment_type in cls._registry:
        raise ValueError(f"支付类型 {payment_type} 已经被注册过了！")
      cls._registry[payment_type] = subclass
      return subclass

    return decorator

  @classmethod
  def create_payment(cls, payment_type: str) -> Payment:
    """核心工厂方法：根据配置或参数动态返回实例"""
    payment_class = cls._registry.get(payment_type)
    if not payment_class:
      raise ValueError(
          f"系统不支持的支付渠道: '{payment_type}'。当前支持的渠道有: {list(cls._registry.keys())}"
      )
    # 实例化并返回具体产品
    return payment_class()


# ==================== 3. 自动化注册（向注册表注入产品） ====================
# 通过装饰器，直接将产品类注册进企业工厂，无需编写任何额外的 XxxFactory 子类！
@EnterprisePaymentFactory.register("wechat")
class RegisteredWeChatPayment(WeChatPayment):
  pass


@EnterprisePaymentFactory.register("alipay")
class RegisteredAliPayPayment(AliPayPayment):
  pass


@EnterprisePaymentFactory.register("paypal")
class RegisteredPayPalPayment(PayPalPayment):
  pass


# ==================== 4. 客户端模拟调用 ====================
if __name__ == "__main__":
  # 模拟从数据库、API请求或配置文件中读取到的用户选择
  user_selections = ["wechat", "alipay", "paypal", "bitcoin"]

  print("=== 开始执行企业级动态多态支付流程 ===")
  for channel in user_selections:
    try:
      print(f"\n正在处理来自外部的渠道请求: [{channel}]")
      # 1. 工厂动态创建对象
      payment_service = EnterprisePaymentFactory.create_payment(channel)
      # 2. 统一接口多态调用
      payment_service.pay(99.8)
    except ValueError as e:
      print(f"❌ 业务异常拦截: {e}")

为什么这是企业级的？

1. 零子类工厂膨胀（Zero Factory Subclasses）：我们彻底废弃了传统的“一个产品对应一个工厂子类”的硬编码，利用 @register 装饰器和 _registry 字典，把 10 几个工厂类精简到了只有一个中心工厂类。
2. 极强的扩展性：如果要新增 ApplePay，只需写一个类并加一行 @EnterprisePaymentFactory.register("apple") 即可，主架构代码一行都不用改。
3. 健壮的错误防御：当客户端传入非法的渠道名称（如代码中的 "bitcoin"）时，工厂能够通过统一的异常处理给出清晰的日志，方便对接上层 API 的错误码返回。

---

五、 总结与架构避坑指南

维度	标准工厂方法模式	企业级注册表工厂模式
代码量	较多（类多、文件多）	极简（核心工厂仅十多行）
扩展新产品	需新增产品类 + 新增工厂子类	仅需新增产品类并使用装饰器注册
配置文件支持	较难直接绑定字符串配置	天生完美支持（字符串直接映射）
适用场景	产品类型固定、层级较深的业务框架	互联网高频迭代、多渠道适配的 SaaS 系统

💡 避坑建议

1. 不要过度设计：如果你的系统里某种对象只有一种实现，且未来 1 年内完全没有扩展需求，直接 obj = MyClass() 实例化是最清晰、最高效的，强行套用工厂方法只会增加阅读代码的心智负担。
2. 多态优于分支：工厂方法的核心目的不是消灭 if-else（那是策略模式干的事），而是消灭 new 对象的硬编码，将对象的生产和消费进行解耦。