Python OOP 设计思想 15:显式是一种设计责任

from dataclasses import dataclassfrom typing import List, Tuple@dataclassclass OrderItem:    """订单项：明确的数据结构定义"""    unit_price: float    quantity: int@dataclassclass Order:    """订单：业务概念的显式建模"""    items: List[OrderItem]    customer_id: strclass OrderTotalCalculator:    """订单总价计算器：专注计算逻辑"""    STANDARD_TAX_RATE = 0.08    def calculate_subtotal(self, order: Order) -> float:        """计算税前总额"""        return sum(            item.unit_price * item.quantity             for item in order.items        )    def calculate_total(self, order: Order) -> float:        """计算含税总额"""        subtotal = self.calculate_subtotal(order)        tax = subtotal * self.STANDARD_TAX_RATE        return subtotal + tax

class DuplicateFinder:    """利用语言特性实现的巧妙但晦涩的重复项查找器"""    def find(self, items):        seen = set()        return [x for x in items if x in seen or seen.add(x)]

class DuplicateFinder:    """明确表达业务逻辑的重复项查找器"""    def find_duplicates(self, items: list) -> list:        """        返回列表中所有重复出现的元素        参数:            items: 待检查的列表        返回:            重复元素的列表，保持首次重复的顺序        """        seen = set()          # 记录已出现的元素        duplicates = []       # 记录发现的重复项        for item in items:            if item in seen:                duplicates.append(item)            else:                seen.add(item)        return duplicates    def find_first_duplicate(self, items: list):        """查找第一个重复出现的元素"""        seen = set()        for item in items:            if item in seen:                return item            seen.add(item)        return None

class DataReader:    """依赖隐式接口的数据读取器"""    def read(self, source):        # 隐含假设：source 必须有 read() 方法        # 隐含假设：read() 返回字符串        # 隐含假设：不会抛出意外异常        return source.read()

from typing import Protocol, runtime_checkablefrom pathlib import Path@runtime_checkableclass Readable(Protocol):    """可读对象的显式接口契约"""    def read(self) -> str:        """读取内容并返回字符串"""        ...class DataReader:    """基于显式契约的数据读取器"""    def read_from(self, source: Readable) -> str:        """        从可读对象读取数据        参数:            source: 必须实现 Readable 协议的对象        返回:            读取到的字符串内容        异常:            TypeError: 如果 source 不满足 Readable 协议            IOError: 由具体实现的 read() 方法抛出        """        # 运行时类型检查（可选）        if not isinstance(source, Readable):            raise TypeError(f"{type(source)} 必须实现 Readable 协议")        return source.read()# 具体实现class FileReader:    """文件读取器：明确实现 Readable 协议"""    def __init__(self, filepath: Path):        self.filepath = filepath    def read(self) -> str:        with open(self.filepath, 'r', encoding='utf-8') as f:            return f.read()class StringReader:    """字符串读取器：适配器模式实现 Readable"""    def __init__(self, content: str):        self.content = content    def read(self) -> str:        return self.content

class DataProcessor:    """充满隐式假设的数据处理器"""    def process(self, data):        # 假设1：data 是字典        # 假设2：data 有 "value" 键        # 假设3："value" 是数值类型        # 假设4：乘以2是有意义的操作        return data["value"] * 2

from typing import TypedDict, NotRequiredclass ProcessData(TypedDict):    """处理数据的明确结构定义"""    value: float    unit: NotRequired[str]  # 可选字段    timestamp: NotRequired[int]class DataProcessor:    """具有明确接口和防御性检查的数据处理器"""    def process(self, data: ProcessData) -> float:        """        处理数据：将数值乘以2        参数:            data: 必须包含 'value' 键，值为数值类型        返回:            处理后的数值        异常:            KeyError: 如果 data 缺少 'value' 键            TypeError: 如果 value 不是数值类型            ValueError: 如果 value 超出合理范围        """        # 输入验证：尽早发现错误        if "value" not in data:            raise KeyError("data 必须包含 'value' 键")        value = data["value"]        # 类型检查        if not isinstance(value, (int, float)):            raise TypeError(f"value 必须是数值类型，得到 {type(value)}")        # 业务逻辑验证        if value < 0:            raise ValueError("value 不能为负数")        # 核心业务逻辑        result = value * 2        # 输出验证（可选）        if not isinstance(result, (int, float)):            raise RuntimeError("计算结果是意外类型")        return result    def process_safe(self, data: ProcessData, default: float = 0.0) -> float:        """安全的处理版本：遇到错误时返回默认值"""        try:            return self.process(data)        except (KeyError, TypeError, ValueError):            # 记录日志            import logging            logging.warning(f"处理数据失败：{data}", exc_info=True)            return default

# 好的命名：自解释的业务语义class OrderPaymentProcessor:    def process_refund_for_cancelled_order(self, order_id: str) -> RefundResult:    # RefundResult 为领域返回对象，此处略        ...# 对比：模糊的命名class OPP:  # 缩写无意义    def prc(self, oid):  # 参数名无意义        ...

from typing import Optional, Union, Literalfrom datetime import datetimefrom decimal import Decimalclass FinancialTransaction:    """金融交易记录"""    def __init__(        self,        amount: Decimal,    # 金融计算用 Decimal，不是 float        currency: Literal["USD", "EUR", "CNY"],    # 有限的合法值        timestamp: datetime,        description: Optional[str] = None,    # 明确可空    ):        self.amount = amount        self.currency = currency        self.timestamp = timestamp        self.description = description    def to_dict(self) -> dict[str, Union[str, Decimal, datetime]]:        """明确返回类型的结构"""        return {            "amount": str(self.amount),            "currency": self.currency,            "timestamp": self.timestamp.isoformat(),            "description": self.description or "",        }

from enum import Enumclass ValidationErrorType(Enum):    """验证错误类型的明确枚举"""    MISSING_FIELD = "missing_field"    INVALID_TYPE = "invalid_type"    OUT_OF_RANGE = "out_of_range"    FORMAT_ERROR = "format_error"class ValidationError(Exception):    """明确的验证异常类"""    def __init__(self, error_type: ValidationErrorType, field: str, message: str):        self.error_type = error_type        self.field = field        self.message = message        super().__init__(f"{field}: {message} ({error_type.value})")class UserValidator:    """用户数据验证器"""    def validate_email(self, email: str) -> str:        """验证邮箱地址，返回规范化的邮箱"""        if not email:            raise ValidationError(                ValidationErrorType.MISSING_FIELD,                "email",                "邮箱地址不能为空"            )        if "@" not in email:            raise ValidationError(                ValidationErrorType.FORMAT_ERROR,                "email",                "邮箱地址格式无效"            )        return email.strip().lower()

from dataclasses import dataclassfrom enum import Enumclass Environment(Enum):    """运行环境枚举"""    DEVELOPMENT = "development"    TESTING = "testing"    STAGING = "staging"    PRODUCTION = "production"class LogLevel(Enum):    """日志级别枚举"""    DEBUG = 10    INFO = 20    WARNING = 30    ERROR = 40    CRITICAL = 50@dataclass(frozen=True)    # 不可变配置class AppConfig:    """应用程序配置：集中管理所有配置项"""    environment: Environment    log_level: LogLevel = LogLevel.INFO    database_url: str = ""    api_timeout_seconds: int = 30    max_retries: int = 3    @classmethod    def from_env(cls) -> "AppConfig":        """从环境变量加载配置"""        import os        env = Environment(os.getenv("APP_ENV", "development"))        return cls(            environment=env,            log_level=LogLevel(int(os.getenv("LOG_LEVEL", 20))),            database_url=os.getenv("DATABASE_URL", ""),            api_timeout_seconds=int(os.getenv("API_TIMEOUT", "30")),            max_retries=int(os.getenv("MAX_RETRIES", "3")),        )    def is_production(self) -> bool:        """明确的判断方法，而不是在代码中直接比较字符串"""        return self.environment == Environment.PRODUCTION

from abc import ABC, abstractmethodfrom typing import List# 仓储层接口class UserRepository(ABC):    """用户数据访问接口"""    @abstractmethod    def get_by_id(self, user_id: str) -> "User":        """根据ID获取用户"""    @abstractmethod    def search(self, query: str, limit: int = 10) -> List["User"]:        """搜索用户"""    @abstractmethod    def save(self, user: "User") -> None:        """保存用户"""    @abstractmethod    def delete(self, user_id: str) -> bool:        """删除用户"""# 领域层class User:    """用户领域对象"""    def __init__(self, user_id: str, email: str, name: str):        self.user_id = user_id        self.email = email        self.name = name    def change_email(self, new_email: str) -> None:        """修改邮箱：领域逻辑"""        if "@" not in new_email:            raise ValueError("无效的邮箱地址")        self.email = new_email# 应用层class UserService:    """用户应用服务"""    def __init__(self, user_repo: UserRepository):        self.user_repo = user_repo    def update_user_email(self, user_id: str, new_email: str) -> None:        """更新用户邮箱：应用逻辑"""        user = self.user_repo.get_by_id(user_id)        user.change_email(new_email)        self.user_repo.save(user)