告别屎山代码！10个Python设计模式让你写出可维护的大项目
数
云朵君约 26 分钟
云朵君约 26 分钟架构决定了代码的寿命，而模式决定了架构的质量
你是否曾回头审视自己半年前写的Python项目，然后心想：这堆垃圾是谁写的？哦等等…好像是我自己。
别担心，每个真正的开发者都有这样的时刻。问题不在于你的编程技巧，而在于你的架构。而架构，正是90%技术债务的诞生地。
经过四年多构建从“可爱小脚本”到“为什么这东西要吃12GB内存？”的系统后，我学到了一个痛苦的教训：语法拯救不了项目，模式才能。
今天，云朵君要分享给你10个Python模式，在构建任何像样的“大项目”之前，你应该把它们刻在脑子里。
这些不是你在每个YouTube教程里看到的基础模式。这些是我多么希望有人在我写下4万行意大利面条式代码之前，就甩在我脸上的模式。
引言：为什么你需要这些模式？
想象一下这个场景：你的项目从简单的脚本成长为有多个模块、需要团队协作的完整应用。突然之间：
添加新功能需要修改5个不同的文件 测试变得异常困难 简单的修改引发意想不到的连锁反应 新同事需要一周才能理解你的代码结构
这就是缺乏设计模式的代价。而下面的10个模式，正是解决这些问题的钥匙。
1. 依赖注入：告别硬编码的紧耦合
大多数开发者不用依赖注入，因为他们觉得“Python是动态语言嘛”。确实…直到你尝试测试一个自己实例化数据库连接的类。
为什么重要？
实现真正的解耦 便于单元测试（可以轻松模拟依赖） 更容易替换实现而不重写逻辑
# 错误的做法：硬编码依赖
classBadUserManager:
def__init__(self):
self.email_service = EmailService() # 硬编码！
defregister(self, user):
# 业务逻辑
self.email_service.send(user, "Welcome!")
# 正确的做法：依赖注入
classEmailService:
defsend(self, to, message):
print(f"发送邮件给 {to}: {message}")
classUserManager:
def__init__(self, email_service):# 依赖作为参数传入
self.email_service = email_service
defregister(self, user):
# 业务逻辑
self.email_service.send(user, "欢迎！")
# 使用
email_service = EmailService()
manager = UserManager(email_service) # 注入依赖
manager.register("[email protected]")
# 测试时可以轻松模拟
classMockEmailService:
defsend(self, to, message):
print(f"[测试] 模拟发送给 {to}")
test_manager = UserManager(MockEmailService())
test_manager.register("[email protected]")
实战技巧：在FastAPI或Django中，依赖注入常用于数据库会话、配置等。试试看，你的测试代码会变得多么干净！
2. 策略模式：告别if/else的丛林
如果你的代码有一个巨大的决策树，恭喜你——你创造了一个怪物。策略模式用清晰、可互换的行为替换了这团乱麻。
# 常见反模式：if/else地狱
defprocess_payment(amount, payment_method):
if payment_method == "paypal":
# 处理PayPal支付
pass
elif payment_method == "stripe":
# 处理Stripe支付
pass
elif payment_method == "alipay":
# 处理支付宝支付
pass
# ... 更多elif
# 优雅的解决方案：策略模式
from abc import ABC, abstractmethod
classPaymentStrategy(ABC):
"""支付策略抽象基类"""
@abstractmethod
defpay(self, amount):
pass
classPayPalPayment(PaymentStrategy):
defpay(self, amount):
print(f"通过PayPal支付 {amount} 元")
# 实际的PayPal API调用
returnTrue
classStripePayment(PaymentStrategy):
defpay(self, amount):
print(f"通过Stripe支付 {amount} 元")
# 实际的Stripe API调用
returnTrue
classAlipayPayment(PaymentStrategy):
defpay(self, amount):
print(f"通过支付宝支付 {amount} 元")
# 实际的支付宝API调用
returnTrue
classPaymentProcessor:
def__init__(self):
self.strategies = {
"paypal": PayPalPayment(),
"stripe": StripePayment(),
"alipay": AlipayPayment()
}
defprocess(self, method, amount):
strategy = self.strategies.get(method)
ifnot strategy:
raise ValueError(f"不支持的支付方式: {method}")
return strategy.pay(amount)
# 使用
processor = PaymentProcessor()
processor.process("paypal", 100)
processor.process("alipay", 200)
# 添加新支付方式非常容易
classWeChatPayment(PaymentStrategy):
defpay(self, amount):
print(f"通过微信支付 {amount} 元")
returnTrue
processor.strategies["wechat"] = WeChatPayment()
设计优势：新增支付方式只需添加新类，无需修改现有代码——这符合开闭原则。
3. 建造者模式：当构造函数变得臃肿时
曾经有一个类有九个可选参数吗？是的。那不是构造函数——那是人质劫持现场。
# 问题：太多参数的构造函数
classUser:
def__init__(self, name, email=None, phone=None, age=None,
address=None, city=None, country=None,
is_active=True, is_admin=False):
# 初始化所有属性...
pass
# 使用起来很痛苦
user = User(
name="张三",
email="[email protected]",
phone="13800138000",
age=25,
address="某街道123号",
city="北京",
country="中国",
is_active=True,
is_admin=False
)
# 解决方案：建造者模式
classQueryBuilder:
def__init__(self, table="users"):
self.table = table
self._select_fields = ["*"]
self._where_conditions = []
self._order_by_fields = []
self._limit_value = None
defselect(self, *fields):
self._select_fields = fields
return self
defwhere(self, condition):
self._where_conditions.append(condition)
return self
deforder_by(self, field, descending=False):
direction = "DESC"if descending else"ASC"
self._order_by_fields.append(f"{field}{direction}")
return self
deflimit(self, count):
self._limit_value = count
return self
defbuild(self):
# 构建SELECT部分
select_clause = f"SELECT {', '.join(self._select_fields)} FROM {self.table}"
# 构建WHERE部分
where_clause = ""
if self._where_conditions:
where_clause = " WHERE " + " AND ".join(self._where_conditions)
# 构建ORDER BY部分
order_clause = ""
if self._order_by_fields:
order_clause = " ORDER BY " + ", ".join(self._order_by_fields)
# 构建LIMIT部分
limit_clause = ""
if self._limit_value:
limit_clause = f" LIMIT {self._limit_value}"
return select_clause + where_clause + order_clause + limit_clause
# 流畅的API调用
query = (
QueryBuilder("users")
.select("id", "name", "email")
.where("age > 18")
.where("is_active = TRUE")
.order_by("created_at", descending=True)
.limit(10)
.build()
)
print(query)
# 输出: SELECT id, name, email FROM users WHERE age > 18 AND is_active = TRUE ORDER BY created_at DESC LIMIT 10
现代应用：SQLAlchemy的查询接口、Django的Q对象、Pydantic的模型配置都使用了类似的建造者模式。
4. 事件驱动模式：当应用需要应对高并发时
大型系统崩溃往往因为一切依赖于一切。解决方案是：事件，而不是链式调用。
classEventBus:
"""事件总线 - 组件通信的中枢"""
def__init__(self):
self._subscribers = {}
defsubscribe(self, event_type, callback):
"""订阅事件"""
if event_type notin self._subscribers:
self._subscribers[event_type] = []
if callback notin self._subscribers[event_type]:
self._subscribers[event_type].append(callback)
defunsubscribe(self, event_type, callback):
"""取消订阅"""
if event_type in self._subscribers:
self._subscribers[event_type].remove(callback)
defpublish(self, event_type, data=None):
"""发布事件"""
if event_type in self._subscribers:
for callback in self._subscribers[event_type]:
try:
callback(data)
except Exception as e:
print(f"事件处理出错: {e}")
defget_subscriber_count(self, event_type):
"""获取订阅者数量"""
return len(self._subscribers.get(event_type, []))
# 定义事件类型
USER_REGISTERED = "user_registered"
ORDER_CREATED = "order_created"
PAYMENT_SUCCESS = "payment_success"
# 创建事件总线实例
bus = EventBus()
# 事件处理器
defsend_welcome_email(user_data):
print(f"[邮件服务] 发送欢迎邮件给 {user_data.get('email')}")
definit_user_storage(user_data):
print(f"[存储服务] 为用户 {user_data.get('username')} 初始化存储空间")
deflog_user_registration(user_data):
print(f"[日志服务] 用户注册: {user_data}")
# 订阅事件
bus.subscribe(USER_REGISTERED, send_welcome_email)
bus.subscribe(USER_REGISTERED, init_user_storage)
bus.subscribe(USER_REGISTERED, log_user_registration)
# 发布事件
new_user = {
"id": 123,
"username": "zhangsan",
"email": "[email protected]",
"registered_at": "2024-01-01 10:00:00"
}
print("用户注册事件触发...")
bus.publish(USER_REGISTERED, new_user)
print(f"事件处理完成，共有{bus.get_subscriber_count(USER_REGISTERED)}个订阅者")
# 输出:
# 用户注册事件触发...
# [邮件服务] 发送欢迎邮件给 [email protected]
# [存储服务] 为用户 zhangsan 初始化存储空间
# [日志服务] 用户注册: {'id': 123, 'username': 'zhangsan', ...}
# 事件处理完成，共有3个订阅者
应用场景：微服务通信、插件系统、GUI应用的事件处理。这种模式比治疗更能解耦有毒的关系。
5. 仓储模式：阻止SQL泄漏到业务逻辑中
如果SQL查询生活在你的业务逻辑中，那么你正在邀请混乱。
import sqlite3
from contextlib import contextmanager
from typing import List, Optional, Dict, Any
@contextmanager
defget_db_connection():
"""数据库连接上下文管理器"""
conn = sqlite3.connect(":memory:") # 使用内存数据库示例
conn.row_factory = sqlite3.Row # 返回字典-like的行
try:
yield conn
finally:
conn.close()
classUserRepository:
"""用户数据仓储"""
def__init__(self, db_connection):
self.db = db_connection
self._create_table()
def_create_table(self):
"""创建用户表"""
self.db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
email TEXT NOT NULL UNIQUE,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.db.commit()
defget_by_id(self, user_id: int) -> Optional[Dict[str, Any]]:
"""根据ID获取用户"""
cursor = self.db.execute(
"SELECT * FROM users WHERE id = ?",
(user_id,)
)
row = cursor.fetchone()
return dict(row) if row elseNone
defget_by_email(self, email: str) -> Optional[Dict[str, Any]]:
"""根据邮箱获取用户"""
cursor = self.db.execute(
"SELECT * FROM users WHERE email = ?",
(email,)
)
row = cursor.fetchone()
return dict(row) if row elseNone
defget_all(self, limit: int = 100) -> List[Dict[str, Any]]:
"""获取所有用户"""
cursor = self.db.execute(
"SELECT * FROM users LIMIT ?",
(limit,)
)
return [dict(row) for row in cursor.fetchall()]
defadd(self, username: str, email: str, age: int = None) -> int:
"""添加用户"""
cursor = self.db.execute(
"INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
(username, email, age)
)
self.db.commit()
return cursor.lastrowid
defupdate(self, user_id: int, **kwargs) -> bool:
"""更新用户信息"""
ifnot kwargs:
returnFalse
set_clause = ", ".join([f"{k} = ?"for k in kwargs.keys()])
values = list(kwargs.values())
values.append(user_id)
self.db.execute(
f"UPDATE users SET {set_clause} WHERE id = ?",
values
)
self.db.commit()
returnTrue
defdelete(self, user_id: int) -> bool:
"""删除用户"""
cursor = self.db.execute(
"DELETE FROM users WHERE id = ?",
(user_id,)
)
self.db.commit()
return cursor.rowcount > 0
# 使用示例
with get_db_connection() as conn:
repo = UserRepository(conn)
# 添加用户
user_id = repo.add("zhangsan", "[email protected]", 25)
print(f"添加用户成功，ID: {user_id}")
# 查询用户
user = repo.get_by_id(user_id)
print(f"查询用户: {user}")
# 更新用户
repo.update(user_id, age=26, email="[email protected]")
# 再次查询
updated_user = repo.get_by_id(user_id)
print(f"更新后用户: {updated_user}")
# 获取所有用户
all_users = repo.get_all()
print(f"总用户数: {len(all_users)}")
架构优势：更换数据库？只需更新一个文件。这就是架构的力量。
6. 映射器模式：你的数据 ≠ 你的对象
将数据结构映射到Python对象可以给你一致性，并消除"字典末日"问题。
from datetime import datetime
from typing import Optional
classUser:
"""领域模型：用户实体"""
def__init__(self, id: int, username: str, email: str,
age: Optional[int] = None, created_at: Optional[datetime] = None):
self.id = id
self.username = username
self.email = email
self.age = age
self.created_at = created_at or datetime.now()
def__repr__(self):
returnf"User(id={self.id}, username='{self.username}', email='{self.email}')"
defis_adult(self) -> bool:
return self.age isnotNoneand self.age >= 18
defget_display_name(self) -> str:
returnf"{self.username} ({self.email})"
classUserMapper:
"""用户映射器：在领域模型和数据模型之间转换"""
@staticmethod
deffrom_dict(data: dict) -> User:
"""从字典创建User对象"""
return User(
id=data.get("id"),
username=data["username"],
email=data["email"],
age=data.get("age"),
created_at=datetime.fromisoformat(data["created_at"]) if"created_at"in data elseNone
)
@staticmethod
deffrom_database_row(row) -> User:
"""从数据库行创建User对象"""
return User(
id=row["id"],
username=row["username"],
email=row["email"],
age=row["age"],
created_at=datetime.fromisoformat(row["created_at"]) if row["created_at"] elseNone
)
@staticmethod
defto_dict(user: User) -> dict:
"""将User对象转为字典"""
return {
"id": user.id,
"username": user.username,
"email": user.email,
"age": user.age,
"created_at": user.created_at.isoformat() if user.created_at elseNone,
"is_adult": user.is_adult(),
"display_name": user.get_display_name()
}
@staticmethod
defto_database_params(user: User) -> tuple:
"""将User对象转为数据库参数"""
return (
user.username,
user.email,
user.age,
user.created_at.isoformat() if user.created_at elseNone
)
# 示例：从API JSON到领域对象
api_response = {
"id": 123,
"username": "zhangsan",
"email": "[email protected]",
"age": 25,
"created_at": "2024-01-01T10:00:00"
}
# 映射到领域对象
user = UserMapper.from_dict(api_response)
print(f"领域对象: {user}")
print(f"是否成人: {user.is_adult()}")
print(f"显示名称: {user.get_display_name()}")
# 映射回字典
user_dict = UserMapper.to_dict(user)
print(f"回传字典: {user_dict}")
# 示例：从数据库到领域对象
classMockDBRow:
"""模拟数据库行"""
def__init__(self):
self.data = {
"id": 456,
"username": "lisi",
"email": "[email protected]",
"age": 17,
"created_at": "2024-01-02T09:00:00"
}
def__getitem__(self, key):
return self.data[key]
db_row = MockDBRow()
user_from_db = UserMapper.from_database_row(db_row)
print(f"\n从数据库创建的用户: {user_from_db}")
print(f"是否成人: {user_from_db.is_adult()}")
API给你JSON，你的代码值得拥有结构。
7. 管道模式：当任务需要流畅执行时
完美适用于机器学习预处理、ETL，或任何你的逻辑有多个阶段的情况。
from abc import ABC, abstractmethod
from typing import Any, List
classPipelineStep(ABC):
"""管道步骤抽象基类"""
@abstractmethod
defprocess(self, data: Any) -> Any:
pass
def__call__(self, data: Any) -> Any:
return self.process(data)
classTextCleanStep(PipelineStep):
"""文本清洗步骤"""
defprocess(self, text: str) -> str:
# 移除多余空格
text = ' '.join(text.split())
# 转换为小写
text = text.lower()
# 移除特殊字符（简单示例）
text = ''.join(c for c in text if c.isalnum() or c.isspace())
return text
classTokenizeStep(PipelineStep):
"""分词步骤"""
def__init__(self, tokenizer=None):
self.tokenizer = tokenizer or str.split
defprocess(self, text: str) -> List[str]:
return self.tokenizer(text)
classStopwordRemovalStep(PipelineStep):
"""停用词移除步骤"""
def__init__(self, stopwords=None):
self.stopwords = stopwords or {"a", "an", "the", "and", "or", "but", "in", "on", "at"}
defprocess(self, tokens: List[str]) -> List[str]:
return [token for token in tokens if token notin self.stopwords]
classStemmingStep(PipelineStep):
"""词干提取步骤（简化版）"""
defprocess(self, tokens: List[str]) -> List[str]:
# 简化版的词干提取
stemmed = []
for token in tokens:
if token.endswith("ing"):
token = token[:-3]
elif token.endswith("ed"):
token = token[:-2]
elif token.endswith("s"):
token = token[:-1]
stemmed.append(token)
return stemmed
classPipeline:
"""管道：连接多个处理步骤"""
def__init__(self, steps: List[PipelineStep] = None):
self.steps = steps or []
defadd_step(self, step: PipelineStep):
"""添加处理步骤"""
self.steps.append(step)
return self # 支持链式调用
defprocess(self, data: Any) -> Any:
"""处理数据"""
result = data
for step in self.steps:
print(f"执行步骤: {step.__class__.__name__}")
result = step.process(result)
print(f"当前结果: {result}")
return result
def__call__(self, data: Any) -> Any:
return self.process(data)
# 构建文本处理管道
text_pipeline = Pipeline()
text_pipeline.add_step(TextCleanStep())
text_pipeline.add_step(TokenizeStep())
text_pipeline.add_step(StopwordRemovalStep())
text_pipeline.add_step(StemmingStep())
# 处理文本
sample_text = " I am RUNNING in the park and playing with dogs! "
print("原始文本:", sample_text)
print("\n开始管道处理...")
final_result = text_pipeline.process(sample_text)
print(f"\n最终结果: {final_result}")
# 输出:
# 原始文本: I am RUNNING in the park and playing with dogs!
# 开始管道处理...
# 执行步骤: TextCleanStep
# 当前结果: i am running in the park and playing with dogs
# 执行步骤: TokenizeStep
# 当前结果: ['i', 'am', 'running', 'in', 'the', 'park', 'and', 'playing', 'with', 'dogs']
# 执行步骤: StopwordRemovalStep
# 当前结果: ['running', 'park', 'playing', 'dogs']
# 执行步骤: StemmingStep
# 当前结果: ['runn', 'park', 'play', 'dog']
# 最终结果: ['runn', 'park', 'play', 'dog']
# 扩展：可以轻松添加新步骤
classSpellCheckStep(PipelineStep):
"""拼写检查步骤（示例）"""
defprocess(self, tokens: List[str]) -> List[str]:
# 这里可以实现实际的拼写检查逻辑
corrected = []
for token in tokens:
# 简单示例：修正常见拼写错误
if token == "runn":
token = "run"
corrected.append(token)
return corrected
# 添加新步骤到现有管道
text_pipeline.add_step(SpellCheckStep())
new_result = text_pipeline.process("I am runn in the parc")
print(f"\n包含拼写检查的结果: {new_result}")
管道保持你的大脑——和你的项目——整洁有序。
8. 命令模式：撤销、重做和"再来一次"
如果你的项目需要可逆操作，这是你的救星。
from abc import ABC, abstractmethod
from typing import List, Dict, Any
classCommand(ABC):
"""命令抽象基类"""
@abstractmethod
defexecute(self) -> Any:
"""执行命令"""
pass
@abstractmethod
defundo(self) -> Any:
"""撤销命令"""
pass
def__str__(self):
return self.__class__.__name__
classTextEditor:
"""文本编辑器（接收者）"""
def__init__(self):
self.text = ""
self.history: List[str] = []
defadd_text(self, new_text: str):
"""添加文本"""
self.history.append(self.text)
self.text += new_text
print(f"添加文本: '{new_text}'")
print(f"当前文本: '{self.text}'")
defdelete_last_word(self):
"""删除最后一个单词"""
if self.text:
self.history.append(self.text)
words = self.text.split()
if words:
last_word = words[-1]
self.text = ' '.join(words[:-1])
print(f"删除最后一个单词: '{last_word}'")
print(f"当前文本: '{self.text}'")
return last_word
return""
defclear_text(self):
"""清空文本"""
self.history.append(self.text)
self.text = ""
print("清空文本")
defrestore_from_history(self):
"""从历史恢复"""
if self.history:
self.text = self.history.pop()
print(f"恢复到: '{self.text}'")
classAddTextCommand(Command):
"""添加文本命令"""
def__init__(self, editor: TextEditor, text: str):
self.editor = editor
self.text = text
self.previous_text = ""
defexecute(self) -> None:
self.previous_text = self.editor.text
self.editor.add_text(self.text)
defundo(self) -> None:
self.editor.text = self.previous_text
print(f"撤销添加文本，恢复为: '{self.editor.text}'")
classDeleteLastWordCommand(Command):
"""删除最后一个单词命令"""
def__init__(self, editor: TextEditor):
self.editor = editor
self.deleted_word = ""
self.previous_text = ""
defexecute(self) -> None:
self.previous_text = self.editor.text
self.deleted_word = self.editor.delete_last_word()
defundo(self) -> None:
self.editor.text = self.previous_text
print(f"撤销删除，恢复为: '{self.editor.text}'")
classClearTextCommand(Command):
"""清空文本命令"""
def__init__(self, editor: TextEditor):
self.editor = editor
self.previous_text = ""
defexecute(self) -> None:
self.previous_text = self.editor.text
self.editor.clear_text()
defundo(self) -> None:
self.editor.text = self.previous_text
print(f"撤销清空，恢复为: '{self.editor.text}'")
classCommandHistory:
"""命令历史管理器"""
def__init__(self):
self.history: List[Command] = []
self.undo_history: List[Command] = []
defexecute_command(self, command: Command):
"""执行命令并记录"""
command.execute()
self.history.append(command)
self.undo_history.clear() # 新的执行清除重做历史
print(f"执行命令: {command}")
defundo(self):
"""撤销上一个命令"""
if self.history:
command = self.history.pop()
command.undo()
self.undo_history.append(command)
print(f"撤销命令: {command}")
else:
print("没有可撤销的命令")
defredo(self):
"""重做上一个撤销的命令"""
if self.undo_history:
command = self.undo_history.pop()
command.execute()
self.history.append(command)
print(f"重做命令: {command}")
else:
print("没有可重做的命令")
defget_history(self):
"""获取命令历史"""
return [str(cmd) for cmd in self.history]
# 使用示例
editor = TextEditor()
history = CommandHistory()
print("=== 文本编辑器演示 ===\n")
# 执行一系列命令
add_hello = AddTextCommand(editor, "Hello ")
history.execute_command(add_hello)
add_world = AddTextCommand(editor, "World ")
history.execute_command(add_world)
add_python = AddTextCommand(editor, "Python ")
history.execute_command(add_python)
print(f"\n当前文本: '{editor.text}'")
print(f"命令历史: {history.get_history()}")
# 撤销操作
print("\n--- 执行撤销 ---")
history.undo()
print(f"撤销后文本: '{editor.text}'")
history.undo()
print(f"再撤销后文本: '{editor.text}'")
# 重做操作
print("\n--- 执行重做 ---")
history.redo()
print(f"重做后文本: '{editor.text}'")
# 执行删除命令
print("\n--- 执行删除 ---")
delete_cmd = DeleteLastWordCommand(editor)
history.execute_command(delete_cmd)
# 执行清空命令
print("\n--- 执行清空 ---")
clear_cmd = ClearTextCommand(editor)
history.execute_command(clear_cmd)
# 多次撤销
print("\n--- 多次撤销 ---")
for _ in range(3):
history.undo()
print(f"\n最终文本: '{editor.text}'")
print(f"最终历史: {history.get_history()}")
UI应用、CLI工具、编辑器——这种模式无处不在。
9. 规范模式：不会腐烂的过滤逻辑
当过滤器变得复杂时，规范模式保持其优雅。
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from datetime import datetime, timedelta
classSpecification(ABC):
"""规范抽象基类"""
@abstractmethod
defis_satisfied(self, item: Dict[str, Any]) -> bool:
"""检查项是否满足规范"""
pass
def__and__(self, other: 'Specification') -> 'AndSpecification':
"""与操作"""
return AndSpecification(self, other)
def__or__(self, other: 'Specification') -> 'OrSpecification':
"""或操作"""
return OrSpecification(self, other)
def__invert__(self) -> 'NotSpecification':
"""非操作"""
return NotSpecification(self)
def__call__(self, item: Dict[str, Any]) -> bool:
"""使规范可调用"""
return self.is_satisfied(item)
classAgeAboveSpecification(Specification):
"""年龄大于规范"""
def__init__(self, min_age: int):
self.min_age = min_age
defis_satisfied(self, user: Dict[str, Any]) -> bool:
age = user.get('age')
return age isnotNoneand age > self.min_age
def__str__(self):
returnf"Age > {self.min_age}"
classHasEmailSpecification(Specification):
"""有邮箱规范"""
defis_satisfied(self, user: Dict[str, Any]) -> bool:
email = user.get('email')
return bool(email and'@'in email)
def__str__(self):
return"Has Email"
classIsActiveSpecification(Specification):
"""活跃用户规范"""
def__init__(self, days_threshold: int = 30):
self.days_threshold = days_threshold
defis_satisfied(self, user: Dict[str, Any]) -> bool:
last_login = user.get('last_login')
ifnot last_login:
returnFalse
if isinstance(last_login, str):
last_login = datetime.fromisoformat(last_login)
days_since_login = (datetime.now() - last_login).days
return days_since_login <= self.days_threshold
def__str__(self):
returnf"Active within {self.days_threshold} days"
classBalanceAboveSpecification(Specification):
"""余额大于规范"""
def__init__(self, min_balance: float):
self.min_balance = min_balance
defis_satisfied(self, user: Dict[str, Any]) -> bool:
balance = user.get('balance', 0)
return balance >= self.min_balance
def__str__(self):
returnf"Balance >= {self.min_balance}"
classAndSpecification(Specification):
"""与规范"""
def__init__(self, *specs: Specification):
self.specs = specs
defis_satisfied(self, item: Dict[str, Any]) -> bool:
return all(spec(item) for spec in self.specs)
def__str__(self):
return" AND ".join(str(spec) for spec in self.specs)
classOrSpecification(Specification):
"""或规范"""
def__init__(self, *specs: Specification):
self.specs = specs
defis_satisfied(self, item: Dict[str, Any]) -> bool:
return any(spec(item) for spec in self.specs)
def__str__(self):
return" OR ".join(str(spec) for spec in self.specs)
classNotSpecification(Specification):
"""非规范"""
def__init__(self, spec: Specification):
self.spec = spec
defis_satisfied(self, item: Dict[str, Any]) -> bool:
returnnot self.spec(item)
def__str__(self):
returnf"NOT ({self.spec})"
classUserFilter:
"""用户过滤器"""
@staticmethod
deffilter_users(users: List[Dict[str, Any]], spec: Specification) -> List[Dict[str, Any]]:
"""根据规范过滤用户"""
return [user for user in users if spec.is_satisfied(user)]
@staticmethod
defcount_users(users: List[Dict[str, Any]], spec: Specification) -> int:
"""统计满足规范的用户数"""
return sum(1for user in users if spec.is_satisfied(user))
# 示例数据
users = [
{"id": 1, "name": "张三", "age": 25, "email": "[email protected]",
"balance": 1000.0, "last_login": "2024-01-15T10:00:00", "is_active": True},
{"id": 2, "name": "李四", "age": 17, "email": "[email protected]",
"balance": 500.0, "last_login": "2023-11-01T09:00:00", "is_active": True},
{"id": 3, "name": "王五", "age": 30, "email": "",
"balance": 2000.0, "last_login": "2024-01-20T14:00:00", "is_active": False},
{"id": 4, "name": "赵六", "age": 22, "email": "[email protected]",
"balance": 300.0, "last_login": "2024-01-18T16:00:00", "is_active": True},
{"id": 5, "name": "孙七", "age": 35, "email": "[email protected]",
"balance": 1500.0, "last_login": "2023-12-01T08:00:00", "is_active": True},
]
print("=== 用户过滤系统 ===\n")
# 基本规范
adult_spec = AgeAboveSpecification(18)
has_email_spec = HasEmailSpecification()
active_spec = IsActiveSpecification(days_threshold=30)
rich_spec = BalanceAboveSpecification(1000)
print("1. 成年用户:")
adults = UserFilter.filter_users(users, adult_spec)
print(f" 规范: {adult_spec}")
print(f" 数量: {len(adults)}")
for user in adults:
print(f" - {user['name']} ({user['age']}岁)")
print("\n2. 有邮箱的活跃成年用户:")
complex_spec = adult_spec & has_email_spec & active_spec
filtered = UserFilter.filter_users(users, complex_spec)
print(f" 规范: {complex_spec}")
print(f" 数量: {len(filtered)}")
for user in filtered:
print(f" - {user['name']}")
print("\n3. 有钱或活跃的用户:")
rich_or_active = rich_spec | active_spec
filtered = UserFilter.filter_users(users, rich_or_active)
print(f" 规范: {rich_or_active}")
print(f" 数量: {len(filtered)}")
for user in filtered:
print(f" - {user['name']} (余额: {user['balance']}, 最近登录: {user['last_login'][:10]})")
print("\n4. 非活跃用户:")
not_active = ~active_spec
filtered = UserFilter.filter_users(users, not_active)
print(f" 规范: {not_active}")
print(f" 数量: {len(filtered)}")
for user in filtered:
print(f" - {user['name']} (最后登录: {user['last_login'][:10]})")
# 动态构建复杂查询
print("\n5. 动态构建的复杂查询:")
# 查找：成年、有邮箱、余额>500，且（是活跃用户或余额>1500）
dynamic_spec = (
adult_spec &
has_email_spec &
BalanceAboveSpecification(500) &
(active_spec | BalanceAboveSpecification(1500))
)
print(f" 规范: {dynamic_spec}")
count = UserFilter.count_users(users, dynamic_spec)
print(f" 满足条件的用户数: {count}")
filtered = UserFilter.filter_users(users, dynamic_spec)
for user in filtered:
print(f" - {user['name']}: 年龄{user['age']}, 余额{user['balance']}, 活跃{user['is_active']}")
向迷宫般的过滤逻辑说再见。
10. 注册表模式：动态插件的智能方式
如果你的"插件系统"使用if name == 'plugin_a':，那么你并没有插件系统。
from abc import ABC, abstractmethod
from typing import Dict, Any, Callable, Optional, Type
import importlib
import pkgutil
classPlugin(ABC):
"""插件基类"""
@abstractmethod
defexecute(self, *args, **kwargs) -> Any:
"""执行插件"""
pass
defget_name(self) -> str:
"""获取插件名称"""
return self.__class__.__name__
defget_version(self) -> str:
"""获取插件版本"""
return"1.0.0"
classPluginRegistry:
"""插件注册表"""
_plugins: Dict[str, Type[Plugin]] = {}
_instances: Dict[str, Plugin] = {}
@classmethod
defregister(cls, name: Optional[str] = None):
"""注册插件装饰器"""
defdecorator(plugin_cls: Type[Plugin]) -> Type[Plugin]:
plugin_name = name or plugin_cls.__name__.lower()
if plugin_name in cls._plugins:
raise ValueError(f"插件 '{plugin_name}' 已注册")
cls._plugins[plugin_name] = plugin_cls
print(f"注册插件: {plugin_name} -> {plugin_cls.__name__}")
return plugin_cls
return decorator
@classmethod
defget_plugin(cls, name: str) -> Optional[Plugin]:
"""获取插件实例"""
if name notin cls._instances:
if name in cls._plugins:
cls._instances[name] = cls._plugins[name]()
else:
returnNone
return cls._instances.get(name)
@classmethod
defget_all_plugins(cls) -> Dict[str, Plugin]:
"""获取所有插件实例"""
# 确保所有插件都已实例化
for name in cls._plugins:
if name notin cls._instances:
cls._instances[name] = cls._plugins[name]()
return cls._instances.copy()
@classmethod
deflist_plugins(cls) -> list:
"""列出所有注册的插件"""
return list(cls._plugins.keys())
@classmethod
defexecute_plugin(cls, name: str, *args, **kwargs) -> Any:
"""执行指定插件"""
plugin = cls.get_plugin(name)
ifnot plugin:
raise ValueError(f"未找到插件: {name}")
print(f"执行插件: {name}")
return plugin.execute(*args, **kwargs)
@classmethod
defload_plugins_from_package(cls, package_name: str):
"""从包中自动加载插件"""
try:
package = importlib.import_module(package_name)
# 遍历包中的所有模块
for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__):
ifnot is_pkg:
full_module_name = f"{package_name}.{module_name}"
importlib.import_module(full_module_name)
print(f"从包 '{package_name}' 加载插件完成")
except ImportError as e:
print(f"加载包 '{package_name}' 失败: {e}")
# 定义一些插件
@PluginRegistry.register("greeter")
classGreeterPlugin(Plugin):
"""问候插件"""
defexecute(self, name: str = "World") -> str:
greeting = f"Hello, {name}!"
print(greeting)
return greeting
defget_version(self) -> str:
return"2.0.0"
@PluginRegistry.register("calculator")
classCalculatorPlugin(Plugin):
"""计算器插件"""
defexecute(self, operation: str, a: float, b: float) -> float:
operations = {
"add": lambda x, y: x + y,
"subtract": lambda x, y: x - y,
"multiply": lambda x, y: x * y,
"divide": lambda x, y: x / y if y != 0else float('inf')
}
if operation notin operations:
raise ValueError(f"不支持的操作: {operation}")
result = operations[operation](a, b)
print(f"{a}{operation}{b} = {result}")
return result
@PluginRegistry.register("uppercase")
classUppercasePlugin(Plugin):
"""大写转换插件"""
defexecute(self, text: str) -> str:
result = text.upper()
print(f"大写转换: '{text}' -> '{result}'")
return result
@PluginRegistry.register() # 使用默认名称
classReversePlugin(Plugin):
"""字符串反转插件"""
defexecute(self, text: str) -> str:
result = text[::-1]
print(f"反转: '{text}' -> '{result}'")
return result
# 动态插件注册
defregister_dynamic_plugin():
"""动态注册插件（例如从配置文件加载）"""
classDynamicPlugin(Plugin):
def__init__(self, name, func):
self._name = name
self._func = func
defexecute(self, *args, **kwargs):
print(f"执行动态插件: {self._name}")
return self._func(*args, **kwargs)
defget_name(self):
return self._name
# 动态创建和注册插件
defcustom_function(x):
return x * 2
# 动态注册
plugin_name = "dynamic_doubler"
PluginRegistry._plugins[plugin_name] = type(
"DynamicDoubler",
(Plugin,),
{
"execute": lambda self, x: custom_function(x),
"get_name": lambda self: plugin_name
}
)
print(f"动态注册插件: {plugin_name}")
# 使用示例
print("=== 插件系统演示 ===\n")
# 列出所有插件
print("已注册的插件:")
for plugin_name in PluginRegistry.list_plugins():
print(f" - {plugin_name}")
print("\n1. 执行问候插件:")
PluginRegistry.execute_plugin("greeter", "Python开发者")
print("\n2. 执行计算器插件:")
result = PluginRegistry.execute_plugin("calculator", "multiply", 6, 7)
print(f" 结果: {result}")
print("\n3. 执行大写插件:")
PluginRegistry.execute_plugin("uppercase", "hello world")
print("\n4. 执行反转插件:")
PluginRegistry.execute_plugin("reverseplugin", "Python") # 使用默认名称
print("\n5. 动态插件演示:")
register_dynamic_plugin()
# 执行动态插件
if"dynamic_doubler"in PluginRegistry.list_plugins():
result = PluginRegistry.execute_plugin("dynamic_doubler", 21)
print(f" 动态插件结果: {result}")
print("\n6. 获取插件信息:")
greeter = PluginRegistry.get_plugin("greeter")
if greeter:
print(f" 插件名称: {greeter.get_name()}")
print(f" 插件版本: {greeter.get_version()}")
print("\n7. 批量执行所有插件:")
all_plugins = PluginRegistry.get_all_plugins()
for name, plugin in all_plugins.items():
print(f" 执行 {name}: ", end="")
if name == "greeter":
plugin.execute("All Plugins")
elif name == "calculator":
plugin.execute("add", 10, 20)
elif name == "uppercase":
plugin.execute("test")
elif name == "reverseplugin":
plugin.execute("abcd")
elif name == "dynamic_doubler":
result = plugin.execute(15)
print(f"结果: {result}", end="")
print()
# 模拟从包加载插件
print("\n8. 模拟从包加载插件:")
# 注意：这里需要实际创建包结构，此处仅演示概念
print(" 调用 PluginRegistry.load_plugins_from_package('my_plugins')")
print(" (假设my_plugins包中有更多插件)")
这种模式为框架、CLI和可扩展系统提供动力。
写在最后
大多数开发者认为架构是"以后学"的东西。这就是为什么你会在凌晨3点重写整个模块。
真相是什么？
模式是可扩展系统和慢动作灾难之间的区别。
当你开始有意识地使用它们时，你的代码会从"勉强能用"跃升到"这维护性也太好了吧"。
如果你已经用Python编码多年，并希望提升水平——这就是真正成长的开始。
实践建议
从一个小模式开始：不要试图一次性应用所有模式。先从依赖注入或策略模式开始 重构旧代码：选择一个小型旧项目，尝试用这些模式重构部分代码 代码审查中关注模式：在团队代码审查中，讨论设计模式的使用 学习经典设计模式：深入研究GoF的23种设计模式，理解其背后的原则
你在项目中用过哪些设计模式？有没有遇到过特别适合或特别不适合使用设计模式的场景？欢迎在评论区分享你的经验和思考！
如果你觉得这篇文章有帮助，欢迎点赞、收藏、转发，让更多开发者受益。关注我，获取更多Python高级技巧和架构知识！
🏴☠️宝藏级🏴☠️ 原创公众号『数据STUDIO』内容超级硬核。公众号以Python为核心语言，垂直于数据科学领域，包括可戳👉Python｜MySQL｜数据分析｜数据可视化｜机器学习与数据挖掘｜爬虫等，从入门到进阶
长按👇关注- 数据STUDIO -设为星标，干货速递