Python解耦优化案例如何实现代码解耦

wen python案例 2026-06-28 237

本文目录导读：

案例场景：用户订单处理系统
效果对比
测试示例

我来通过一个具体的案例,展示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()