本文目录导读：
我来通过一个具体的案例,展示Python代码解耦的实现方法。
案例场景：用户订单处理系统
原始耦合代码（不推荐）
# tightly_coupled.py - 高耦合版本
class OrderProcessor:
def __init__(self, db_config, email_config):
self.db_config = db_config
self.email_config = email_config
self.db_connection = None
self.email_client = None
def process_order(self, order_data):
# 1. 验证订单
if not self.validate_order(order_data):
return False
# 2. 保存到数据库（直接操作数据库）
import sqlite3
conn = sqlite3.connect(self.db_config['database'])
cursor = conn.cursor()
cursor.execute(
"INSERT INTO orders (user_id, product, amount) VALUES (?, ?, ?)",
(order_data['user_id'], order_data['product'], order_data['amount'])
)
conn.commit()
conn.close()
# 3. 发送邮件（直接调用SMTP）
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(f"订单已处理: {order_data['product']}")
msg['Subject'] = '订单通知'
msg['From'] = self.email_config['from']
msg['To'] = order_data['email']
with smtplib.SMTP(self.email_config['smtp_server']) as server:
server.login(self.email_config['user'], self.email_config['password'])
server.send_message(msg)
return True
def validate_order(self, order_data):
return all([
order_data.get('user_id'),
order_data.get('product'),
order_data.get('amount', 0) > 0
])
解耦优化版本
# decoupled_version.py - 解耦版本
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Any, Optional
import logging
# 1. 数据类 - 定义数据结构
@dataclass
class Order:
user_id: str
product: str
amount: float
email: str
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Order':
return cls(
user_id=data['user_id'],
product=data['product'],
amount=data['amount'],
email=data['email']
)
# 2. 接口抽象 - 定义契约
class Validator(ABC):
@abstractmethod
def validate(self, order: Order) -> bool:
pass
class Database(ABC):
@abstractmethod
def save(self, order: Order) -> bool:
pass
@abstractmethod
def connect(self) -> bool:
pass
@abstractmethod
def disconnect(self):
pass
class Notifier(ABC):
@abstractmethod
def send(self, order: Order, message: str) -> bool:
pass
# 3. 具体实现 - 每个功能独立实现
class OrderValidator(Validator):
def validate(self, order: Order) -> bool:
if not order.user_id:
logging.error("User ID is missing")
return False
if not order.product:
logging.error("Product is missing")
return False
if order.amount <= 0:
logging.error(f"Invalid amount: {order.amount}")
return False
if not order.email:
logging.error("Email is missing")
return False
return True
class SQLiteDatabase(Database):
def __init__(self, db_path: str):
self.db_path = db_path
self.connection = None
def connect(self) -> bool:
try:
import sqlite3
self.connection = sqlite3.connect(self.db_path)
self._create_table()
return True
except Exception as e:
logging.error(f"Database connection failed: {e}")
return False
def _create_table(self):
cursor = self.connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT,
product TEXT,
amount REAL,
email TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.connection.commit()
def save(self, order: Order) -> bool:
try:
cursor = self.connection.cursor()
cursor.execute(
"INSERT INTO orders (user_id, product, amount, email) VALUES (?, ?, ?, ?)",
(order.user_id, order.product, order.amount, order.email)
)
self.connection.commit()
return True
except Exception as e:
logging.error(f"Failed to save order: {e}")
return False
def disconnect(self):
if self.connection:
self.connection.close()
class EmailNotifier(Notifier):
def __init__(self, smtp_config: Dict[str, str]):
self.smtp_config = smtp_config
def send(self, order: Order, message: str) -> bool:
try:
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(message)
msg['Subject'] = '订单通知'
msg['From'] = self.smtp_config['from']
msg['To'] = order.email
with smtplib.SMTP(self.smtp_config['server']) as server:
server.login(self.smtp_config['user'], self.smtp_config['password'])
server.send_message(msg)
return True
except Exception as e:
logging.error(f"Failed to send email: {e}")
return False
# 4. 依赖注入容器
class ServiceContainer:
def __init__(self):
self._services = {}
self._instances = {}
def register(self, name: str, service_class, *args, **kwargs):
self._services[name] = (service_class, args, kwargs)
def get(self, name: str):
if name not in self._instances:
service_class, args, kwargs = self._services[name]
self._instances[name] = service_class(*args, **kwargs)
return self._instances[name]
# 5. 主处理器 - 使用依赖注入
class OrderProcessor:
def __init__(self, validator: Validator, database: Database, notifier: Optional[Notifier] = None):
self.validator = validator
self.database = database
self.notifier = notifier
self.logger = logging.getLogger(__name__)
def process_order(self, order_data: Dict[str, Any]) -> bool:
# 转换数据
order = Order.from_dict(order_data)
# 验证
if not self.validator.validate(order):
return False
# 保存到数据库
if not self.database.save(order):
return False
# 发送通知（可选）
if self.notifier:
message = f"订单已处理: {order.product}, 金额: {order.amount}"
self.notifier.send(order, message)
self.logger.info(f"Order processed successfully: {order.user_id}")
return True
# 6. 配置和初始化
def setup_container():
container = ServiceContainer()
# 注册服务
container.register('validator', OrderValidator)
container.register('database', SQLiteDatabase, 'orders.db')
container.register('notifier', EmailNotifier, {
'server': 'smtp.gmail.com',
'from': 'noreply@example.com',
'user': 'user@example.com',
'password': 'password'
})
return container
# 7. 使用示例
def main():
# 方式1: 手动注入
validator = OrderValidator()
database = SQLiteDatabase('orders.db')
database.connect()
notifier = EmailNotifier({
'server': 'smtp.gmail.com',
'from': 'noreply@example.com',
'user': 'user@example.com',
'password': 'password'
})
processor = OrderProcessor(validator, database, notifier)
# 处理订单
order_data = {
'user_id': '123',
'product': 'Python Book',
'amount': 29.99,
'email': 'user@example.com'
}
result = processor.process_order(order_data)
print(f"Order processed: {result}")
# 方式2: 使用容器
# container = setup_container()
# processor = OrderProcessor(
# container.get('validator'),
# container.get('database'),
# container.get('notifier')
# )
if __name__ == "__main__":
# 配置日志
logging.basicConfig(level=logging.INFO)
main()
效果对比
原始版本问题：
- ❌ 硬编码数据库操作
- ❌ 硬编码邮件发送
- ❌ 难以测试（无法mock）
- ❌ 无法替换实现（如切换数据库）
- ❌ 违反单一职责原则
解耦版本优势：
- ✅ 每个组件职责清晰
- ✅ 易于测试和Mock
- ✅ 可以灵活替换实现
- ✅ 遵循开闭原则
- ✅ 支持依赖注入
- ✅ 配置化初始化
测试示例
# test_order_processor.py
import unittest
from unittest.mock import Mock
from decoupled_version import OrderProcessor, Order
class TestOrderProcessor(unittest.TestCase):
def setUp(self):
# 使用Mock对象
self.mock_validator = Mock()
self.mock_database = Mock()
self.mock_notifier = Mock()
self.processor = OrderProcessor(
self.mock_validator,
self.mock_database,
self.mock_notifier
)
def test_process_order_success(self):
# 配置Mock
self.mock_validator.validate.return_value = True
self.mock_database.save.return_value = True
order_data = {
'user_id': '123',
'product': 'Test Book',
'amount': 29.99,
'email': 'test@example.com'
}
result = self.processor.process_order(order_data)
self.assertTrue(result)
self.mock_validator.validate.assert_called_once()
self.mock_database.save.assert_called_once()
self.mock_notifier.send.assert_called_once()
if __name__ == '__main__':
unittest.main()
- 依赖注入：通过构造函数或setter方法注入依赖
- 接口抽象：使用ABC定义清晰的接口契约
- 单一职责：每个类只负责一个功能
- 配置分离：将配置与代码分离
- 单元测试：解耦后更容易编写单元测试
- 日志替换：使用日志而不是print语句
这样解耦后,你可以轻松替换数据库、通知方式或验证逻辑，而无需修改核心处理逻辑。
