SOLID 原则在 Python 中的实践

SOLID 原则在 Python 中的实践 | 极客日志

# 反例：典型的'上帝类'（God Class）
class BusinessProcessor:
    def __init__(self):
        self.db_connection = None
        self.http_client = None
        self.config = {}

    def validate_user(self, user_data): pass
    def save_to_database(self, user_data): pass
    def send_email(self, user_email): pass
    def generate_report(self, data): pass
    def handle_api_request(self, request): pass

# 违反 SRP 的版本
class UserRegistration:
    """违反 SRP 的用户注册类"""
    def __init__(self):
        self.db = Database()
        self.email_sender = EmailSender()
        self.validator = Validator()

    def register_user(self, user_data):
        # 1. 验证输入
        if not self.validator.validate_email(user_data['email']):
            raise ValueError("Invalid email")
        if not self.validator.validate_password(user_data['password']):
            raise ValueError("Invalid password")
        # 2. 业务逻辑检查
        if self.db.user_exists(user_data['email']):
            raise ValueError("User already exists")
        # 3. 创建用户
        user = self._create_user_object(user_data)
        # 4. 保存到数据库
        self.db.save_user(user)
        # 5. 发送欢迎邮件
        self.email_sender.send_welcome_email(user.email)
        # 6. 记录日志
        self._log_registration(user)
        return user

    def _create_user_object(self, user_data): pass
    def _log_registration(self, user): pass

# 遵守 SRP 的版本
class UserValidator:
    """专职验证"""
    def validate_registration_data(self, user_data):
        if not self.validate_email(user_data['email']):
            return False, "Invalid email"
        if not self.validate_password(user_data['password']):
            return False, "Invalid password"
        return True, "Valid"

    def validate_email(self, email): return "@" in email
    def validate_password(self, password): return len(password) >= 8

class UserRepository:
    """专职数据持久化"""
    def __init__(self, db):
        self.db = db
    def save_user(self, user): return self.db.save_user(user)
    def user_exists(self, email): return self.db.user_exists(email)

class EmailService:
    """专职邮件发送"""
    def send_welcome_email(self, email): pass

class RegistrationLogger:
    """专职日志记录"""
    def log_registration(self, user): pass

class UserRegistrationService:
    """协调用户注册流程，符合 SRP"""
    def __init__(self, validator, repository, email_service, logger):
        self.validator = validator
        self.repository = repository
        self.email_service = email_service
        self.logger = logger

    def register_user(self, user_data):
        is_valid, message = self.validator.validate_registration_data(user_data)
        if not is_valid:
            raise ValueError(message)
        if self.repository.user_exists(user_data['email']):
            raise ValueError("User already exists")
        user = User(**user_data)
        self.repository.save_user(user)
        self.email_service.send_welcome_email(user.email)
        self.logger.log_registration(user)
        return user

# 测试变得简单
def test_user_validator():
    validator = UserValidator()
    is_valid, msg = validator.validate_registration_data({
        'email': '[email protected]',
        'password': 'weak'
    })
    assert not is_valid
    assert "password" in msg

def test_user_registration_service():
    mock_validator = Mock()
    mock_repository = Mock()
    mock_email = Mock()
    mock_logger = Mock()
    service = UserRegistrationService(
        mock_validator, mock_repository, mock_email, mock_logger
    )
    # 测试逻辑...

from abc import ABC, abstractmethod
from typing import List, Dict, Any

# 违反 OCP 的实现
class ReportExporter:
    """违反 OCP 的报表导出器"""
    def export(self, data: List[Dict], format_type: str):
        if format_type == "pdf": return self._export_pdf(data)
        elif format_type == "excel": return self._export_excel(data)
        elif format_type == "csv": return self._export_csv(data)
        else: raise ValueError(f"Unsupported format: {format_type}")

    def _export_pdf(self, data): return f"PDF: {data}"
    def _export_excel(self, data): return f"Excel: {data}"
    def _export_csv(self, data): return f"CSV: {data}"

# 定义抽象接口
class ExportStrategy(ABC):
    """导出策略抽象基类"""
    @abstractmethod
    def export(self, data: List[Dict]) -> str: pass

# 具体策略实现
class PDFExportStrategy(ExportStrategy):
    def export(self, data: List[Dict]) -> str: return f"PDF: {data}"

class ExcelExportStrategy(ExportStrategy):
    def export(self, data: List[Dict]) -> str: return f"Excel: {data}"

class CSVExportStrategy(ExportStrategy):
    def export(self, data: List[Dict]) -> str: return f"CSV: {data}"

class JSONExportStrategy(ExportStrategy):
    def export(self, data: List[Dict]) -> str:
        import json
        return json.dumps(data)

# 上下文类
class ReportExporter:
    """符合 OCP 的报表导出器"""
    def __init__(self):
        self._strategies = {}
        self._register_default_strategies()

    def _register_default_strategies(self):
        self._strategies = {
            "pdf": PDFExportStrategy(),
            "excel": ExcelExportStrategy(),
            "csv": CSVExportStrategy()
        }

    def register_strategy(self, format_type: str, strategy: ExportStrategy):
        self._strategies[format_type] = strategy

    def export(self, data: List[Dict], format_type: str) -> str:
        if format_type not in self._strategies:
            raise ValueError(f"Unsupported format: {format_type}")
        strategy = self._strategies[format_type]
        return strategy.export(data)

# 使用示例
exporter = ReportExporter()
exporter.register_strategy("json", JSONExportStrategy())
data = [{"name": "Alice", "value": 100}, {"name": "Bob", "value": 200}]
pdf_result = exporter.export(data, "pdf")
json_result = exporter.export(data, "json")

from typing import Protocol, List, Dict

class ExportStrategy(Protocol):
    def export(self, data: List[Dict]) -> str: ...

class XMLExportStrategy:
    def export(self, data: List[Dict]) -> str: return f"<data>{data}</data>"

class CustomExporter:
    def __init__(self, strategy: ExportStrategy):
        self.strategy = strategy
    def execute_export(self, data: List[Dict]) -> str:
        return self.strategy.export(data)

exporter = CustomExporter(XMLExportStrategy())
result = exporter.execute_export([{"test": "data"}])

# 使用__slots__减少内存开销
class PDFExportStrategy:
    __slots__ = ['config']
    def __init__(self, config=None):
        self.config = config or {}
    def export(self, data): pass

# 使用函数代替类（轻量级策略）
def pdf_export_strategy(data, config=None):
    return f"PDF: {data}"

class OptimizedReportExporter:
    def __init__(self):
        self._strategies = {
            "pdf": pdf_export_strategy,
        }
    def export(self, data, format_type):
        strategy = self._strategies.get(format_type)
        if strategy is None:
            raise ValueError(f"Unsupported format: {format_type}")
        return strategy(data)

# 违反 DIP 的实现
class MySQLDatabase:
    def connect(self, host, user, password, database): pass
    def query(self, sql): return [{"id": 1, "name": "John"}]
    def close(self): pass

class UserService:
    def __init__(self):
        self.db = MySQLDatabase()
        self.connection = self.db.connect("localhost", "user", "password", "mydb")
    def get_users(self): return self.db.query("SELECT * FROM users")
    def __del__(self): self.db.close()

from abc import ABC, abstractmethod
from typing import List, Dict, Any

# 抽象层
class Database(ABC):
    @abstractmethod
    def connect(self, **kwargs): pass
    @abstractmethod
    def query(self, sql: str) -> List[Dict[str, Any]]: pass
    @abstractmethod
    def close(self): pass

# 低层模块实现抽象
class MySQLDatabase(Database):
    def connect(self, **kwargs): print("Connecting to MySQL..."); return None
    def query(self, sql: str) -> List[Dict[str, Any]]: print(f"Executing MySQL query: {sql}"); return [{"id": 1, "name": "John"}]
    def close(self): print("Closing MySQL connection")

class PostgreSQLDatabase(Database):
    def connect(self, **kwargs): print("Connecting to PostgreSQL..."); return None
    def query(self, sql: str) -> List[Dict[str, Any]]: print(f"Executing PostgreSQL query: {sql}"); return [{"id": 1, "name": "John", "postgres_specific": True}]
    def close(self): print("Closing PostgreSQL connection")

# 高层模块依赖抽象
class UserService:
    def __init__(self, db: Database):
        self.db = db
        self.connection = self.db.connect()
    def get_users(self) -> List[Dict[str, Any]]: return self.db.query("SELECT * FROM users")
    def get_user_by_id(self, user_id: int) -> Dict[str, Any]:
        result = self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
        return result[0] if result else {}
    def close(self): self.db.close()

class DIContainer:
    def __init__(self):
        self._dependencies = {}
    def register(self, abstract, concrete):
        self._dependencies[abstract] = concrete
    def resolve(self, abstract):
        if abstract not in self._dependencies:
            raise ValueError(f"Dependency not registered: {abstract}")
        return self._dependencies[abstract]()

def configure_dependencies():
    container = DIContainer()
    container.register(Database, MySQLDatabase)
    return container

container = configure_dependencies()
db = container.resolve(Database)
user_service = UserService(db)
users = user_service.get_users()
user_service.close()

# 方法 1：构造函数注入（最常用）
class UserService:
    def __init__(self, db: Database, email_sender: EmailSender):
        self.db = db
        self.email_sender = email_sender

# 方法 2：设置方法注入
class UserService:
    def set_database(self, db: Database): self.db = db
    def set_email_sender(self, email_sender: EmailSender): self.email_sender = email_sender

# 方法 3：上下文管理器注入（Pythonic 方式）
class DatabaseContext:
    def __init__(self, db: Database):
        self.db = db
    def __enter__(self):
        self.connection = self.db.connect()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.db.close()
    def get_users(self): return self.db.query("SELECT * FROM users")

with DatabaseContext(MySQLDatabase()) as db_context:
    users = db_context.get_users()

import functools
from typing import Dict, Any

def inject_dependencies(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        dependencies = resolve_dependencies()
        for dep_name, dep_instance in dependencies.items():
            if dep_name not in kwargs:
                kwargs[dep_name] = dep_instance
        return func(*args, **kwargs)
    return wrapper

def resolve_dependencies() -> Dict[str, Any]:
    return {
        'db': MySQLDatabase(),
        'email_sender': SMTPEmailSender(),
        'logger': FileLogger()
    }

class UserController:
    @inject_dependencies
    def create_user(self, user_data, db=None, email_sender=None, logger=None):
        user = db.save_user(user_data)
        email_sender.send_welcome_email(user.email)
        logger.log(f"User created: {user.id}")
        return user

controller = UserController()
user = controller.create_user({"name": "Alice", "email": "[email protected]"})

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from datetime import datetime
import uuid

class Order:
    def __init__(self, order_id: str, items: List[Dict], total: float):
        self.order_id = order_id
        self.items = items
        self.total = total
        self.status = "created"
        self.created_at = datetime.now()

class OrderValidator(ABC):
    @abstractmethod
    def validate(self, order: Order) -> bool: pass

class InventoryManager(ABC):
    @abstractmethod
    def check_stock(self, product_id: str, quantity: int) -> bool: pass
    @abstractmethod
    def reserve_stock(self, product_id: str, quantity: int) -> bool: pass

class PaymentProcessor(ABC):
    @abstractmethod
    def process_payment(self, order: Order, payment_method: str) -> bool: pass

class Notifier(ABC):
    @abstractmethod
    def send_notification(self, recipient: str, message: str) -> bool: pass

class BasicOrderValidator(OrderValidator):
    def validate(self, order: Order) -> bool:
        if not order.items: return False, "Order must have items"
        if order.total <= 0: return False, "Order total must be positive"
        return True, "Valid"

class DatabaseInventoryManager(InventoryManager):
    def check_stock(self, product_id: str, quantity: int) -> bool: return True
    def reserve_stock(self, product_id: str, quantity: int) -> bool: return True

class StripePaymentProcessor(PaymentProcessor):
    def process_payment(self, order: Order, payment_method: str) -> bool:
        print(f"Processing payment of ${order.total} via Stripe")
        return True

class EmailNotifier(Notifier):
    def send_notification(self, recipient: str, message: str) -> bool:
        print(f"Sending email to {recipient}: {message}")
        return True

class OrderService:
    def __init__(self, validator: OrderValidator, inventory_manager: InventoryManager, payment_processor: PaymentProcessor, notifier: Notifier):
        self.validator = validator
        self.inventory_manager = inventory_manager
        self.payment_processor = payment_processor
        self.notifier = notifier

    def create_order(self, order_data: Dict, payment_method: str) -> Order:
        order = Order(order_id=str(uuid.uuid4()), items=order_data['items'], total=order_data['total'])
        is_valid, message = self.validator.validate(order)
        if not is_valid: raise ValueError(f"Order validation failed: {message}")
        for item in order.items:
            if not self.inventory_manager.check_stock(item['product_id'], item['quantity']):
                raise ValueError(f"Insufficient stock for product {item['product_id']}")
        for item in order.items:
            self.inventory_manager.reserve_stock(item['product_id'], item['quantity'])
        payment_success = self.payment_processor.process_payment(order, payment_method)
        if not payment_success:
            self._release_reserved_stock(order.items)
            raise ValueError("Payment processing failed")
        order.status = "completed"
        self.notifier.send_notification(order_data['customer_email'], f"Order {order.order_id} completed successfully")
        return order

    def _release_reserved_stock(self, items: List[Dict]): pass

def create_order_service() -> OrderService:
    validator = BasicOrderValidator()
    inventory_manager = DatabaseInventoryManager()
    payment_processor = StripePaymentProcessor()
    notifier = EmailNotifier()
    return OrderService(validator, inventory_manager, payment_processor, notifier)

def main():
    order_service = create_order_service()
    order_data = {
        'items': [{'product_id': 'prod1', 'quantity': 2, 'price': 50}, {'product_id': 'prod2', 'quantity': 1, 'price': 30}],
        'total': 130,
        'customer_email': '[email protected]'
    }
    try:
        order = order_service.create_order(order_data, "credit_card")
        print(f"Order created successfully: {order.order_id}")
    except ValueError as e:
        print(f"Order creation failed: {e}")

if __name__ == "__main__":
    main()

# 添加新的支付方式（符合 OCP）
class PayPalPaymentProcessor(PaymentProcessor):
    def process_payment(self, order: Order, payment_method: str) -> bool:
        print(f"Processing payment of ${order.total} via PayPal")
        return True

# 添加新的验证规则（符合 OCP）
class FraudDetectionValidator(OrderValidator):
    def __init__(self, base_validator: OrderValidator):
        self.base_validator = base_validator
    def validate(self, order: Order) -> bool:
        is_valid, message = self.base_validator.validate(order)
        if not is_valid: return False, message
        if order.total > 10000: return False, "Large order requires manual review"
        return True, "Valid"

def create_enhanced_order_service() -> OrderService:
    base_validator = BasicOrderValidator()
    validator = FraudDetectionValidator(base_validator)
    inventory_manager = DatabaseInventoryManager()
    payment_processor = PayPalPaymentProcessor()
    notifier = EmailNotifier()
    return OrderService(validator, inventory_manager, payment_processor, notifier)

class LazyDependency:
    def __init__(self, factory):
        self._factory = factory
        self._instance = None
    def __getattr__(self, name):
        if self._instance is None:
            self._instance = self._factory()
        return getattr(self._instance, name)

class OptimizedOrderService:
    def __init__(self, db_factory, email_factory):
        self._db = LazyDependency(db_factory)
        self._email_sender = LazyDependency(email_factory)
    @property
    def db(self): return self._db
    @property
    def email_sender(self): return self._email_sender

from functools import lru_cache

class CachedOrderService:
    def __init__(self, order_repository):
        self.order_repository = order_repository
    @lru_cache(maxsize=1000)
    def get_order(self, order_id: str) -> Order:
        return self.order_repository.find_by_id(order_id)

# 反例：不必要的抽象
class AbstractOrderCreator(ABC):
    @abstractmethod
    def create(self): pass

class OrderCreator(AbstractOrderCreator):
    def create(self): pass

# 正例：简单函数
def create_order(order_data): pass

class SimpleContainer:
    _instances = {}
    @classmethod
    def register(cls, abstract, implementation):
        cls._instances[abstract] = implementation
    @classmethod
    def resolve(cls, abstract):
        return cls._instances.get(abstract)

SimpleContainer.register(Database, MySQLDatabase)
SimpleContainer.register(EmailSender, SMTPEmailSender)
db = SimpleContainer.resolve(Database)

场景	性能开销	可维护性提升	适用性建议
简单 CRUD 应用	5-10%	30-50%	推荐
高性能计算	15-25%	20-30%	谨慎使用
长期维护系统	3-8%	100-200%	强烈推荐
微服务架构	2-5%	50-100%	推荐

SOLID 原则在 Python 中的实践

SOLID 原则在 Python 中的实践

1 引言：为什么 Python 开发者更需要 SOLID 原则

1.1 Python 的动态特性是一把双刃剑

1.2 SOLID 原则的 Python 化解读

2 单一职责原则（SRP）：专注的力量

2.1 SRP 核心理解：变更的理由是关键

2.2 实战：用户注册流程的重构

2.3 SRP 带来的架构收益

3 开闭原则（OCP）：扩展的艺术

3.1 OCP 本质：用扩展代替修改

3.2 实战：多格式导出系统

3.3 Python 协议：更灵活的 OCP 实现

3.4 OCP 性能优化技巧

4 依赖倒置原则（DIP）：面向抽象编程

4.1 DIP 核心：高层模块不应该依赖低层模块

4.2 实战：数据库访问层的依赖倒置

4.3 依赖注入的 Pythonic 实现

4.4 基于装饰器的依赖注入

5 企业级实战：电子商务订单系统

5.1 需求分析：复杂的业务场景

5.2 基于 SOLID 原则的架构设计

5.3 扩展性演示：添加新功能

6 性能优化与故障排查

6.1 SOLID 架构的性能考量

6.2 常见问题与解决方案

7 总结与最佳实践

7.1 SOLID 原则在 Python 中的特殊考量

7.2 何时应该（和不应该）应用 SOLID

7.3 性能数据总结

官方文档与权威参考

更多推荐文章

相关免费在线工具

SOLID 原则在 Python 中的实践

SOLID 原则在 Python 中的实践

1 引言：为什么 Python 开发者更需要 SOLID 原则

1.1 Python 的动态特性是一把双刃剑

1.2 SOLID 原则的 Python 化解读

2 单一职责原则（SRP）：专注的力量

2.1 SRP 核心理解：变更的理由是关键

2.2 实战：用户注册流程的重构

2.3 SRP 带来的架构收益

3 开闭原则（OCP）：扩展的艺术

3.1 OCP 本质：用扩展代替修改

3.2 实战：多格式导出系统

3.3 Python 协议：更灵活的 OCP 实现

3.4 OCP 性能优化技巧

4 依赖倒置原则（DIP）：面向抽象编程

4.1 DIP 核心：高层模块不应该依赖低层模块

4.2 实战：数据库访问层的依赖倒置

4.3 依赖注入的 Pythonic 实现

4.4 基于装饰器的依赖注入

5 企业级实战：电子商务订单系统

5.1 需求分析：复杂的业务场景

5.2 基于 SOLID 原则的架构设计

5.3 扩展性演示：添加新功能

6 性能优化与故障排查

6.1 SOLID 架构的性能考量

6.2 常见问题与解决方案

7 总结与最佳实践

7.1 SOLID 原则在 Python 中的特殊考量

7.2 何时应该（和不应该）应用 SOLID

7.3 性能数据总结

官方文档与权威参考

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具