Python设计模式实战：用Pythonic的方式实现单例、工厂模式

『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨
写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉 CodeCapsule官网

Python设计模式实战：用Pythonic的方式实现单例、工厂模式

1. 引言

设计模式是软件工程中解决常见问题的经典方案，它们代表了最佳的实践经验。然而，在Python这样具有独特哲学的语言中，传统的设计模式实现方式往往显得过于复杂和"不Pythonic"。Pythonic的代码应该简洁、可读性强，并且充分利用Python的语言特性。

本文将深入探讨如何在Python中优雅地实现单例模式和工厂模式，展示Python特有的实现方式，并提供实际的应用场景示例。我们将避免Java或C++风格的冗长实现，转而采用符合Python哲学的精炼代码。

2. Pythonic编程哲学

2.1 Python之禅（The Zen of Python）

在深入设计模式之前，让我们回顾一下Python的核心哲学：

import this

输出结果体现了Python的设计原则：

• 优美胜于丑陋
• 明了胜于晦涩
• 简单胜于复杂
• 扁平胜于嵌套
• 可读性很重要

2.2 Pythonic代码的特征

Pythonic代码具有以下特点：

• 充分利用Python的内置特性
• 代码简洁但功能强大
• 遵循约定俗成的惯例
• 易于阅读和维护
• 避免不必要的复杂性

3. 单例模式（Singleton Pattern）

3.1 单例模式概述

单例模式确保一个类只有一个实例，并提供一个全局访问点。在Python中，我们有多种方式实现单例，每种方式都有其适用场景。

3.2 模块级别的单例

在Python中，模块本身就是天然的单例，这是最简单也是最Pythonic的实现方式：

# singleton_module.py
"""
模块级别的单例实现
这是最Pythonic的单例实现方式
"""class _DatabaseConnection:"""数据库连接类，通过模块实现单例"""def __init__(self):self.connection_count = 0self.is_connected = Falsedef connect(self):"""模拟数据库连接"""if not self.is_connected:self.connection_count += 1self.is_connected = Trueprint(f"数据库连接已建立 (连接次数: {self.connection_count})")else:print("数据库已经连接")def disconnect(self):"""模拟断开数据库连接"""if self.is_connected:self.is_connected = Falseprint("数据库连接已断开")else:print("数据库未连接")def execute_query(self, query):"""执行查询"""if self.is_connected:print(f"执行查询: {query}")return f"查询结果: {query}"else:raise ConnectionError("数据库未连接")# 创建单例实例
database = _DatabaseConnection()# 使用示例
if __name__ == "__main__":# 在另一个模块中，只需要导入database即可使用单例from singleton_module import databasedatabase.connect()database.execute_query("SELECT * FROM users")database.disconnect()

3.3 使用元类实现单例

元类提供了更灵活的单例控制方式：

class SingletonMeta(type):"""单例元类使用元类控制类的创建过程，确保只有一个实例"""_instances = {}def __call__(cls, *args, **kwargs):"""当类被调用（实例化）时触发"""if cls not in cls._instances:instance = super().__call__(*args, **kwargs)cls._instances[cls] = instancereturn cls._instances[cls]class ConfigurationManager(metaclass=SingletonMeta):"""配置管理器单例使用元类实现单例模式"""def __init__(self):self._config = {}self._load_default_config()def _load_default_config(self):"""加载默认配置"""self._config = {'app_name': 'My Application','version': '1.0.0','debug': False,'database_url': 'sqlite:///default.db','max_connections': 10}def get(self, key, default=None):"""获取配置值"""return self._config.get(key, default)def set(self, key, value):"""设置配置值"""self._config[key] = valuedef update(self, **kwargs):"""批量更新配置"""self._config.update(kwargs)def __getitem__(self, key):"""支持下标访问"""return self._config[key]def __setitem__(self, key, value):"""支持下标设置"""self._config[key] = valuedef __contains__(self, key):"""支持in操作符"""return key in self._configdef __str__(self):"""字符串表示"""return f"ConfigurationManager(config={self._config})"# 使用示例
def test_configuration_singleton():"""测试配置管理器单例"""config1 = ConfigurationManager()config2 = ConfigurationManager()print(f"config1 is config2: {config1 is config2}")  # 输出: True# 设置配置config1.set('debug', True)config1['timeout'] = 30# 在另一个实例中访问print(f"config2 debug mode: {config2.get('debug')}")  # 输出: Trueprint(f"config2 timeout: {config2['timeout']}")  # 输出: 30# 测试包含检查print(f"'version' in config1: {'version' in config1}")  # 输出: True

3.4 使用装饰器实现单例

装饰器提供了另一种优雅的单例实现方式：

from functools import wraps
from threading import Lockdef singleton(cls):"""单例装饰器使用装饰器实现线程安全的单例模式"""instances = {}lock = Lock()@wraps(cls)def get_instance(*args, **kwargs):with lock:if cls not in instances:instances[cls] = cls(*args, **kwargs)return instances[cls]return get_instancedef singleton_with_parameters(*singleton_args, **singleton_kwargs):"""带参数的单例装饰器支持根据不同的参数创建不同的单例实例"""def decorator(cls):instances = {}lock = Lock()@wraps(cls)def get_instance(*args, **kwargs):# 根据装饰器参数和实例化参数生成唯一键key = (singleton_args, tuple(sorted(singleton_kwargs.items())), args, tuple(sorted(kwargs.items())))with lock:if key not in instances:instances[key] = cls(*args, **kwargs)return instances[key]return get_instancereturn decorator@singleton
class Logger:"""日志管理器单例使用装饰器实现单例模式"""def __init__(self):self._logs = []self._log_level = 'INFO'def set_level(self, level):"""设置日志级别"""self._log_level = level.upper()def log(self, message, level='INFO'):"""记录日志"""timestamp = self._get_timestamp()log_entry = {'timestamp': timestamp,'level': level.upper(),'message': message}self._logs.append(log_entry)if self._should_log(level):print(f"[{level.upper()}] {timestamp}: {message}")def _get_timestamp(self):"""获取时间戳"""from datetime import datetimereturn datetime.now().isoformat()def _should_log(self, level):"""判断是否应该记录日志"""levels = {'DEBUG': 0, 'INFO': 1, 'WARNING': 2, 'ERROR': 3, 'CRITICAL': 4}current_level = levels.get(self._log_level, 1)message_level = levels.get(level.upper(), 1)return message_level >= current_leveldef get_logs(self, level=None):"""获取日志记录"""if level:return [log for log in self._logs if log['level'] == level.upper()]return self._logs.copy()def clear_logs(self):"""清空日志"""self._logs.clear()@singleton_with_parameters(scope='application')
class CacheManager:"""缓存管理器使用带参数的装饰器实现单例"""def __init__(self, name="default"):self.name = nameself._cache = {}self._hits = 0self._misses = 0def set(self, key, value, ttl=None):"""设置缓存值"""import timeexpire_time = time.time() + ttl if ttl else Noneself._cache[key] = {'value': value,'expire_time': expire_time}def get(self, key):"""获取缓存值"""import timeif key in self._cache:item = self._cache[key]if item['expire_time'] is None or item['expire_time'] > time.time():self._hits += 1return item['value']else:# 过期删除del self._cache[key]self._misses += 1return Nonedef delete(self, key):"""删除缓存值"""if key in self._cache:del self._cache[key]return Truereturn Falsedef clear(self):"""清空缓存"""self._cache.clear()self._hits = 0self._misses = 0def stats(self):"""获取缓存统计"""hit_rate = self._hits / (self._hits + self._misses) if (self._hits + self._misses) > 0 else 0return {'hits': self._hits,'misses': self._misses,'hit_rate': hit_rate,'size': len(self._cache)}

3.5 单例模式的最佳实践

class PythonicSingleton:"""Pythonic单例模式最佳实践结合了多种Python特性的单例实现"""_instance = None_initialized = Falsedef __new__(cls, *args, **kwargs):"""控制实例创建"""if cls._instance is None:cls._instance = super().__new__(cls)return cls._instancedef __init__(self):"""确保只初始化一次"""if not self._initialized:self._initialize()self._initialized = Truedef _initialize(self):"""初始化方法"""self.data = {}self.initialization_time = self._get_current_time()def _get_current_time(self):"""获取当前时间"""from datetime import datetimereturn datetime.now()def __repr__(self):return f"<PythonicSingleton at {id(self)}>"@classmethoddef get_instance(cls):"""获取单例实例的类方法"""if cls._instance is None:cls()return cls._instance

4. 工厂模式（Factory Pattern）

4.1 工厂模式概述

工厂模式提供了一种创建对象的方式，而无需指定具体的类。在Python中，我们可以利用动态语言的特性实现更加灵活的工厂模式。

4.2 简单工厂模式

from enum import Enum
from typing import Dict, Type, Anyclass NotificationType(Enum):"""通知类型枚举"""EMAIL = "email"SMS = "sms"PUSH = "push"SLACK = "slack"class Notification:"""通知基类"""def __init__(self, recipient: str, message: str):self.recipient = recipientself.message = messagedef send(self) -> bool:"""发送通知"""raise NotImplementedError("子类必须实现send方法")def validate(self) -> bool:"""验证通知参数"""return bool(self.recipient and self.message)class EmailNotification(Notification):"""邮件通知"""def send(self) -> bool:print(f"发送邮件给 {self.recipient}: {self.message}")# 模拟发送逻辑return Truedef __str__(self):return f"EmailNotification(to={self.recipient})"class SMSNotification(Notification):"""短信通知"""def send(self) -> bool:print(f"发送短信给 {self.recipient}: {self.message}")# 模拟发送逻辑return Truedef __str__(self):return f"SMSNotification(to={self.recipient})"class PushNotification(Notification):"""推送通知"""def send(self) -> bool:print(f"发送推送给 {self.recipient}: {self.message}")# 模拟发送逻辑return Truedef __str__(self):return f"PushNotification(to={self.recipient})"class NotificationFactory:"""通知工厂类简单工厂模式实现"""# 注册的通知类型映射_notification_types: Dict[NotificationType, Type[Notification]] = {NotificationType.EMAIL: EmailNotification,NotificationType.SMS: SMSNotification,NotificationType.PUSH: PushNotification,}@classmethoddef create_notification(cls, notification_type: NotificationType,recipient: str, message: str,**kwargs) -> Notification:"""创建通知实例Args:notification_type: 通知类型recipient: 接收者message: 消息内容**kwargs: 额外参数Returns:Notification实例"""if notification_type not in cls._notification_types:raise ValueError(f"不支持的通知类型: {notification_type}")notification_class = cls._notification_types[notification_type]return notification_class(recipient, message, **kwargs)@classmethoddef register_notification_type(cls, notification_type: NotificationType,notification_class: Type[Notification]):"""注册新的通知类型Args:notification_type: 通知类型notification_class: 通知类"""if not issubclass(notification_class, Notification):raise TypeError("通知类必须继承自Notification")cls._notification_types[notification_type] = notification_classprint(f"已注册通知类型: {notification_type} -> {notification_class.__name__}")@classmethoddef list_notification_types(cls) -> list:"""列出所有支持的通知类型"""return list(cls._notification_types.keys())# 使用示例
def test_notification_factory():"""测试通知工厂"""# 创建不同类型的通知email_notification = NotificationFactory.create_notification(NotificationType.EMAIL, "[e-mail]", "欢迎使用我们的服务！")sms_notification = NotificationFactory.create_notification(NotificationType.SMS,"[phone]","您的验证码是: 123456")# 发送通知notifications = [email_notification, sms_notification]for notification in notifications:if notification.validate():success = notification.send()print(f"{notification} 发送{'成功' if success else '失败'}")

4.3 工厂方法模式

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List@dataclass
class Document:"""文档数据类"""title: strcontent: strformat: strdef display(self):"""显示文档信息"""print(f"文档: {self.title}")print(f"格式: {self.format}")print(f"内容: {self.content[:50]}..." if len(self.content) > 50 else f"内容: {self.content}")class DocumentExporter(ABC):"""文档导出器抽象基类工厂方法模式"""def __init__(self, document: Document):self.document = document@abstractmethoddef export(self) -> str:"""导出文档"""passdef pre_process(self) -> None:"""预处理文档"""print(f"预处理文档: {self.document.title}")def post_process(self, exported_content: str) -> str:"""后处理导出内容"""print(f"后处理导出内容")return exported_contentdef export_document(self) -> str:"""导出文档的模板方法"""self.pre_process()content = self.export()return self.post_process(content)class PDFExporter(DocumentExporter):"""PDF导出器"""def export(self) -> str:"""导出为PDF格式"""print(f"将文档 '{self.document.title}' 导出为PDF")return f"PDF内容: {self.document.content}"class WordExporter(DocumentExporter):"""Word文档导出器"""def export(self) -> str:"""导出为Word格式"""print(f"将文档 '{self.document.title}' 导出为Word文档")return f"Word内容: {self.document.content}"class HTMLExporter(DocumentExporter):"""HTML导出器"""def export(self) -> str:"""导出为HTML格式"""print(f"将文档 '{self.document.title}' 导出为HTML")html_content = f"""<!DOCTYPE html><html><head><title>{self.document.title}</title></head><body><h1>{self.document.title}</h1><p>{self.document.content}</p></body></html>"""return html_contentclass DocumentExportFactory:"""文档导出工厂工厂方法模式实现"""_exporters = {'pdf': PDFExporter,'word': WordExporter,'html': HTMLExporter}@classmethoddef create_exporter(cls, format_type: str, document: Document) -> DocumentExporter:"""创建文档导出器Args:format_type: 导出格式document: 文档对象Returns:文档导出器实例"""if format_type not in cls._exporters:raise ValueError(f"不支持的导出格式: {format_type}")exporter_class = cls._exporters[format_type]return exporter_class(document)@classmethoddef register_exporter(cls, format_type: str, exporter_class: type):"""注册新的导出器"""if not issubclass(exporter_class, DocumentExporter):raise TypeError("导出器必须继承自DocumentExporter")cls._exporters[format_type] = exporter_classprint(f"已注册导出器: {format_type} -> {exporter_class.__name__}")@classmethoddef get_supported_formats(cls) -> List[str]:"""获取支持的导出格式"""return list(cls._exporters.keys())# 使用示例
def test_document_export():"""测试文档导出"""# 创建文档document = Document(title="Python设计模式指南",content="本文介绍Python中的设计模式实现...",format="markdown")# 导出为不同格式formats = ['pdf', 'word', 'html']for format_type in formats:print(f"\n--- 导出为 {format_type.upper()} ---")try:exporter = DocumentExportFactory.create_exporter(format_type, document)result = exporter.export_document()print(f"导出结果: {result[:100]}...")except ValueError as e:print(f"导出失败: {e}")

4.4 抽象工厂模式

from abc import ABC, abstractmethod
from enum import Enum
from typing import Protocolclass Theme(Enum):"""主题枚举"""LIGHT = "light"DARK = "dark"BLUE = "blue"class Button(Protocol):"""按钮接口"""def render(self) -> str:"""渲染按钮"""...def on_click(self) -> str:"""点击事件"""...class Checkbox(Protocol):"""复选框接口"""def render(self) -> str:"""渲染复选框"""...def toggle(self) -> str:"""切换状态"""...class LightButton:"""浅色主题按钮"""def render(self) -> str:return "渲染浅色按钮"def on_click(self) -> str:return "浅色按钮被点击"class DarkButton:"""深色主题按钮"""def render(self) -> str:return "渲染深色按钮"def on_click(self) -> str:return "深色按钮被点击"class BlueButton:"""蓝色主题按钮"""def render(self) -> str:return "渲染蓝色按钮"def on_click(self) -> str:return "蓝色按钮被点击"class LightCheckbox:"""浅色主题复选框"""def __init__(self):self.checked = Falsedef render(self) -> str:state = "选中" if self.checked else "未选中"return f"浅色复选框 [{state}]"def toggle(self) -> str:self.checked = not self.checkedstate = "选中" if self.checked else "未选中"return f"浅色复选框状态: {state}"class DarkCheckbox:"""深色主题复选框"""def __init__(self):self.checked = Falsedef render(self) -> str:state = "选中" if self.checked else "未选中"return f"深色复选框 [{state}]"def toggle(self) -> str:self.checked = not self.checkedstate = "选中" if self.checked else "未选中"return f"深色复选框状态: {state}"class BlueCheckbox:"""蓝色主题复选框"""def __init__(self):self.checked = Falsedef render(self) -> str:state = "选中" if self.checked else "未选中"return f"蓝色复选框 [{state}]"def toggle(self) -> str:self.checked = not self.checkedstate = "选中" if self.checked else "未选中"return f"蓝色复选框状态: {state}"class GUIFactory(ABC):"""GUI工厂抽象基类抽象工厂模式"""@abstractmethoddef create_button(self) -> Button:"""创建按钮"""pass@abstractmethoddef create_checkbox(self) -> Checkbox:"""创建复选框"""passclass LightThemeFactory(GUIFactory):"""浅色主题工厂"""def create_button(self) -> Button:return LightButton()def create_checkbox(self) -> Checkbox:return LightCheckbox()class DarkThemeFactory(GUIFactory):"""深色主题工厂"""def create_button(self) -> Button:return DarkButton()def create_checkbox(self) -> Checkbox:return DarkCheckbox()class BlueThemeFactory(GUIFactory):"""蓝色主题工厂"""def create_button(self) -> Button:return BlueButton()def create_checkbox(self) -> Checkbox:return BlueCheckbox()class ThemeFactory:"""主题工厂抽象工厂的工厂"""_factories = {Theme.LIGHT: LightThemeFactory,Theme.DARK: DarkThemeFactory,Theme.BLUE: BlueThemeFactory}@classmethoddef create_factory(cls, theme: Theme) -> GUIFactory:"""创建主题工厂Args:theme: 主题类型Returns:GUI工厂实例"""if theme not in cls._factories:raise ValueError(f"不支持的主题: {theme}")factory_class = cls._factories[theme]return factory_class()@classmethoddef register_theme(cls, theme: Theme, factory_class: type):"""注册新主题"""if not issubclass(factory_class, GUIFactory):raise TypeError("工厂类必须继承自GUIFactory")cls._factories[theme] = factory_classprint(f"已注册主题: {theme} -> {factory_class.__name__}")class Application:"""应用程序类使用抽象工厂创建UI组件"""def __init__(self, theme: Theme):self.theme = themeself.factory = ThemeFactory.create_factory(theme)self.button = Noneself.checkbox = Nonedef create_ui(self):"""创建用户界面"""self.button = self.factory.create_button()self.checkbox = self.factory.create_checkbox()print(f"使用 {self.theme.value} 主题创建UI")print(f"按钮: {self.button.render()}")print(f"复选框: {self.checkbox.render()}")def simulate_interaction(self):"""模拟用户交互"""if self.button and self.checkbox:print(f"\n用户交互:")print(self.button.on_click())print(self.checkbox.toggle())print(self.checkbox.toggle())# 使用示例
def test_abstract_factory():"""测试抽象工厂"""themes = [Theme.LIGHT, Theme.DARK, Theme.BLUE]for theme in themes:print(f"\n{'='*50}")print(f"测试 {theme.value.upper()} 主题")print('='*50)app = Application(theme)app.create_ui()app.simulate_interaction()

5. 完整代码示例

下面是一个完整的示例，展示了如何在真实场景中结合使用单例模式和工厂模式：

#!/usr/bin/env python3
"""
Python设计模式实战完整示例
结合单例模式和工厂模式的完整应用
"""import json
import sqlite3
from abc import ABC, abstractmethod
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Any
from pathlib import Path
import logging# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class StorageType(Enum):"""存储类型枚举"""MEMORY = "memory"SQLITE = "sqlite"JSON_FILE = "json_file"@dataclass
class User:"""用户数据类"""id: intname: stremail: strcreated_at: strupdated_at: strclass UserRepository(ABC):"""用户仓库抽象基类"""@abstractmethoddef save(self, user: User) -> bool:"""保存用户"""pass@abstractmethoddef find_by_id(self, user_id: int) -> Optional[User]:"""根据ID查找用户"""pass@abstractmethoddef find_all(self) -> List[User]:"""查找所有用户"""pass@abstractmethoddef delete(self, user_id: int) -> bool:"""删除用户"""passclass MemoryUserRepository(UserRepository):"""内存用户仓库"""def __init__(self):self._users: Dict[int, User] = {}self._next_id = 1def save(self, user: User) -> bool:if user.id == 0:  # 新用户user.id = self._next_idself._next_id += 1user.updated_at = datetime.now().isoformat()self._users[user.id] = userlogger.info(f"用户已保存到内存: {user}")return Truedef find_by_id(self, user_id: int) -> Optional[User]:return self._users.get(user_id)def find_all(self) -> List[User]:return list(self._users.values())def delete(self, user_id: int) -> bool:if user_id in self._users:del self._users[user_id]logger.info(f"用户已从内存删除: {user_id}")return Truereturn Falseclass SQLiteUserRepository(UserRepository):"""SQLite用户仓库"""def __init__(self, db_path: str = "users.db"):self.db_path = db_pathself._init_database()def _init_database(self):"""初始化数据库"""with sqlite3.connect(self.db_path) as conn:conn.execute('''CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT,name TEXT NOT NULL,email TEXT UNIQUE NOT NULL,created_at TEXT NOT NULL,updated_at TEXT NOT NULL)''')conn.commit()def save(self, user: User) -> bool:with sqlite3.connect(self.db_path) as conn:cursor = conn.cursor()if user.id == 0:  # 新用户user.created_at = datetime.now().isoformat()user.updated_at = user.created_atcursor.execute('''INSERT INTO users (name, email, created_at, updated_at)VALUES (?, ?, ?, ?)''', (user.name, user.email, user.created_at, user.updated_at))user.id = cursor.lastrowidelse:  # 更新用户user.updated_at = datetime.now().isoformat()cursor.execute('''UPDATE users SET name = ?, email = ?, updated_at = ?WHERE id = ?''', (user.name, user.email, user.updated_at, user.id))conn.commit()logger.info(f"用户已保存到SQLite: {user}")return Truedef find_by_id(self, user_id: int) -> Optional[User]:with sqlite3.connect(self.db_path) as conn:cursor = conn.cursor()cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))row = cursor.fetchone()if row:return User(*row)return Nonedef find_all(self) -> List[User]:with sqlite3.connect(self.db_path) as conn:cursor = conn.cursor()cursor.execute('SELECT * FROM users')rows = cursor.fetchall()return [User(*row) for row in rows]def delete(self, user_id: int) -> bool:with sqlite3.connect(self.db_path) as conn:cursor = conn.cursor()cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))conn.commit()deleted = cursor.rowcount > 0if deleted:logger.info(f"用户已从SQLite删除: {user_id}")return deletedclass JsonFileUserRepository(UserRepository):"""JSON文件用户仓库"""def __init__(self, file_path: str = "users.json"):self.file_path = Path(file_path)self._ensure_file_exists()def _ensure_file_exists(self):"""确保文件存在"""if not self.file_path.exists():self.file_path.write_text('[]')def _read_users(self) -> List[User]:"""读取用户数据"""try:data = json.loads(self.file_path.read_text())return [User(**user_data) for user_data in data]except (json.JSONDecodeError, KeyError):return []def _write_users(self, users: List[User]):"""写入用户数据"""data = [asdict(user) for user in users]self.file_path.write_text(json.dumps(data, indent=2))def save(self, user: User) -> bool:users = self._read_users()if user.id == 0:  # 新用户user.id = max([u.id for u in users], default=0) + 1user.created_at = datetime.now().isoformat()user.updated_at = user.created_atusers.append(user)else:  # 更新用户for i, existing_user in enumerate(users):if existing_user.id == user.id:user.updated_at = datetime.now().isoformat()users[i] = userbreakself._write_users(users)logger.info(f"用户已保存到JSON文件: {user}")return Truedef find_by_id(self, user_id: int) -> Optional[User]:users = self._read_users()for user in users:if user.id == user_id:return userreturn Nonedef find_all(self) -> List[User]:return self._read_users()def delete(self, user_id: int) -> bool:users = self._read_users()initial_count = len(users)users = [user for user in users if user.id != user_id]if len(users) < initial_count:self._write_users(users)logger.info(f"用户已从JSON文件删除: {user_id}")return Truereturn Falseclass RepositoryFactory:"""仓库工厂"""_repositories = {StorageType.MEMORY: MemoryUserRepository,StorageType.SQLITE: SQLiteUserRepository,StorageType.JSON_FILE: JsonFileUserRepository}@classmethoddef create_repository(cls, storage_type: StorageType, **kwargs) -> UserRepository:"""创建用户仓库"""if storage_type not in cls._repositories:raise ValueError(f"不支持的存储类型: {storage_type}")repository_class = cls._repositories[storage_type]return repository_class(**kwargs)@classmethoddef register_repository(cls, storage_type: StorageType, repository_class: type):"""注册新的仓库类型"""if not issubclass(repository_class, UserRepository):raise TypeError("仓库类必须继承自UserRepository")cls._repositories[storage_type] = repository_classlogger.info(f"已注册存储类型: {storage_type} -> {repository_class.__name__}")class UserService:"""用户服务类使用单例模式和工厂模式"""_instance = Nonedef __new__(cls, storage_type: StorageType = StorageType.MEMORY, **kwargs):if cls._instance is None:cls._instance = super().__new__(cls)cls._instance._initialized = Falsereturn cls._instancedef __init__(self, storage_type: StorageType = StorageType.MEMORY, **kwargs):if not self._initialized:self.repository = RepositoryFactory.create_repository(storage_type, **kwargs)self._initialized = Truelogger.info(f"UserService已初始化，使用存储类型: {storage_type}")def create_user(self, name: str, email: str) -> User:"""创建用户"""user = User(id=0,name=name,email=email,created_at="",updated_at="")if self.repository.save(user):return userelse:raise RuntimeError("创建用户失败")def get_user(self, user_id: int) -> Optional[User]:"""获取用户"""return self.repository.find_by_id(user_id)def get_all_users(self) -> List[User]:"""获取所有用户"""return self.repository.find_all()def update_user(self, user_id: int, name: str = None, email: str = None) -> bool:"""更新用户"""user = self.repository.find_by_id(user_id)if not user:return Falseif name:user.name = nameif email:user.email = emailreturn self.repository.save(user)def delete_user(self, user_id: int) -> bool:"""删除用户"""return self.repository.delete(user_id)@classmethoddef get_instance(cls) -> 'UserService':"""获取单例实例"""if cls._instance is None:cls()return cls._instancedef main():"""主函数 - 演示完整的使用场景"""print("=" * 60)print("Python设计模式实战演示")print("单例模式 + 工厂模式完整示例")print("=" * 60)# 测试不同的存储类型storage_types = [(StorageType.MEMORY, {}),(StorageType.SQLITE, {"db_path": "test_users.db"}),(StorageType.JSON_FILE, {"file_path": "test_users.json"})]for storage_type, kwargs in storage_types:print(f"\n{'='*50}")print(f"测试存储类型: {storage_type.value}")print('='*50)# 创建用户服务单例user_service = UserService(storage_type, **kwargs)# 测试单例特性another_instance = UserService.get_instance()print(f"单例验证: user_service is another_instance = {user_service is another_instance}")# 创建用户users_data = [("张三", "[e-mail]"),("李四", "[e-mail]"),("王五", "[e-mail]")]created_users = []for name, email in users_data:user = user_service.create_user(name, email)created_users.append(user)print(f"创建用户: {user.name} (ID: {user.id})")# 查询用户print(f"\n所有用户:")all_users = user_service.get_all_users()for user in all_users:print(f"  - {user.id}: {user.name} <{user.email}>")# 更新用户if created_users:first_user = created_users[0]user_service.update_user(first_user.id, name="张三丰")updated_user = user_service.get_user(first_user.id)print(f"\n更新用户: {updated_user.name} <{updated_user.email}>")# 删除用户if len(created_users) > 1:user_to_delete = created_users[1]user_service.delete_user(user_to_delete.id)print(f"\n删除用户: {user_to_delete.name}")# 验证删除remaining_users = user_service.get_all_users()print(f"剩余用户数: {len(remaining_users)}")print(f"\n{storage_type.value} 存储测试完成")print("\n" + "=" * 60)print("演示完成！")print("=" * 60)if __name__ == "__main__":main()

6. 设计模式的最佳实践

6.1 Pythonic设计原则

1. 简单优于复杂

# 不Pythonic
class ComplexSingleton:_instance = None_lock = threading.Lock()def __new__(cls, *args, **kwargs):with cls._lock:if cls._instance is None:cls._instance = super().__new__(cls)return cls._instance# Pythonic
class SimpleSingleton:_instance = Nonedef __new__(cls):if cls._instance is None:cls._instance = super().__new__(cls)return cls._instance

2. 利用语言特性

# 使用模块作为单例
# database.py
class Database:def connect(self):passdb = Database()# 在其他模块中使用
from database import db

6.2 测试策略

import unittest
from unittest.mock import Mock, patchclass TestSingletonPattern(unittest.TestCase):"""单例模式测试"""def test_singleton_instance(self):"""测试单例实例唯一性"""instance1 = UserService.get_instance()instance2 = UserService.get_instance()self.assertIs(instance1, instance2)self.assertEqual(id(instance1), id(instance2))def test_singleton_initialization(self):"""测试单例初始化"""service = UserService.get_instance()self.assertTrue(hasattr(service, 'repository'))self.assertIsNotNone(service.repository)class TestFactoryPattern(unittest.TestCase):"""工厂模式测试"""def test_repository_factory(self):"""测试仓库工厂"""# 测试内存仓库memory_repo = RepositoryFactory.create_repository(StorageType.MEMORY)self.assertIsInstance(memory_repo, MemoryUserRepository)# 测试SQLite仓库sqlite_repo = RepositoryFactory.create_repository(StorageType.SQLITE)self.assertIsInstance(sqlite_repo, SQLiteUserRepository)def test_factory_registration(self):"""测试工厂注册机制"""class TestRepository(UserRepository):def save(self, user): return Truedef find_by_id(self, user_id): return Nonedef find_all(self): return []def delete(self, user_id): return True# 注册新的仓库类型RepositoryFactory.register_repository(StorageType.MEMORY, TestRepository)# 验证注册成功repo = RepositoryFactory.create_repository(StorageType.MEMORY)self.assertIsInstance(repo, TestRepository)class TestIntegration(unittest.TestCase):"""集成测试"""def setUp(self):"""测试准备"""self.user_service = UserService(StorageType.MEMORY)def test_user_crud_operations(self):"""测试用户CRUD操作"""# 创建用户user = self.user_service.create_user("测试用户", "[e-mail]")self.assertIsNotNone(user.id)# 读取用户retrieved_user = self.user_service.get_user(user.id)self.assertEqual(user.id, retrieved_user.id)# 更新用户self.user_service.update_user(user.id, name="更新后的用户")updated_user = self.user_service.get_user(user.id)self.assertEqual(updated_user.name, "更新后的用户")# 删除用户delete_result = self.user_service.delete_user(user.id)self.assertTrue(delete_result)# 验证删除deleted_user = self.user_service.get_user(user.id)self.assertIsNone(deleted_user)if __name__ == '__main__':unittest.main()

7. 性能考虑与优化

7.1 单例模式的性能优化

import time
from functools import lru_cache
from threading import RLockclass OptimizedSingleton:"""优化的单例实现结合了多种性能优化技术"""_instance = None_lock = RLock()  # 可重入锁，性能更好def __init__(self):# 延迟初始化昂贵资源self._expensive_resource = Nonedef __new__(cls):# 双重检查锁定，减少锁竞争if cls._instance is None:with cls._lock:if cls._instance is None:cls._instance = super().__new__(cls)return cls._instance@propertydef expensive_resource(self):"""延迟加载昂贵资源"""if self._expensive_resource is None:with self._lock:if self._expensive_resource is None:self._expensive_resource = self._load_expensive_resource()return self._expensive_resourcedef _load_expensive_resource(self):"""模拟加载昂贵资源"""time.sleep(0.1)  # 模拟耗时操作return "昂贵的资源数据"@lru_cache(maxsize=128)def cached_operation(self, param):"""缓存昂贵操作的结果"""time.sleep(0.05)  # 模拟耗时操作return f"操作结果: {param}"

7.2 工厂模式的性能优化

from functools import lru_cacheclass OptimizedFactory:"""优化的工厂实现使用缓存提高性能"""@classmethod@lru_cache(maxsize=128)def create_cached_instance(cls, class_name, *args, **kwargs):"""创建缓存实例"""# 根据类名创建实例并缓存import importlibmodule_name, class_name = class_name.rsplit('.', 1)module = importlib.import_module(module_name)class_ = getattr(module, class_name)return class_(*args, **kwargs)class FastRepositoryFactory:"""快速仓库工厂使用预注册和缓存"""_repository_cache = {}@classmethoddef create_repository(cls, storage_type: StorageType, **kwargs) -> UserRepository:# 使用缓存键cache_key = (storage_type, tuple(sorted(kwargs.items())))if cache_key not in cls._repository_cache:repository = RepositoryFactory.create_repository(storage_type, **kwargs)cls._repository_cache[cache_key] = repositoryreturn cls._repository_cache[cache_key]

8. 总结

通过本文的深入探讨，我们学习了如何在Python中优雅地实现单例模式和工厂模式。关键要点总结如下：

8.1 单例模式要点

1. 模块单例是最Pythonic的实现方式
2. 装饰器提供灵活的单例控制
3. 元类适合需要精细控制类创建的场景
4. 考虑线程安全和性能优化

8.2 工厂模式要点

1. 简单工厂适合对象创建逻辑不复杂的场景
2. 工厂方法通过子类化提供扩展性
3. 抽象工厂用于创建相关对象家族
4. 利用Python的动态特性实现灵活的工厂

8.3 Pythonic设计原则

• 充分利用Python语言特性
• 保持代码简洁和可读性
• 遵循"显式优于隐式"的原则
• 适当使用装饰器、元类等高级特性

8.4 实际应用建议

1. 评估需求：选择最适合问题场景的模式
2. 保持简单：避免过度设计
3. 考虑测试：设计可测试的代码结构
4. 性能考量：在性能敏感的场景进行优化

设计模式是强大的工具，但在Python中，我们应该始终以Pythonic的方式使用它们。通过结合Python的语言特性和设计模式的精髓，我们可以写出既优雅又实用的代码。

记住，最好的设计模式实现是那些既解决了问题，又保持了代码简洁性和可读性的实现。在Python的世界里，简单往往就是最好的解决方案。