Python重构工具案例如何封装代码重构

wen python案例 2026-06-28 236

本文目录导读：

函数提取与封装
数据验证封装
策略模式重构
数据管道封装
重构工具建议

我来分享几个Python代码重构的实用案例,重点介绍如何封装重构逻辑。

函数提取与封装

原始代码

# 混乱的业务逻辑
def process_order(order_data):
    # 验证订单
    if not order_data.get('user_id'):
        raise ValueError("用户ID不能为空")
    if not order_data.get('items'):
        raise ValueError("订单项不能为空")
    if len(order_data['items']) == 0:
        raise ValueError("订单项不能为空")
    # 计算价格
    total = 0
    for item in order_data['items']:
        price = item['price'] * item['quantity']
        total += price
    # 应用折扣
    if total > 1000:
        total = total * 0.9
    elif total > 500:
        total = total * 0.95
    # 记录日志
    print(f"订单处理完成，总价: {total}")
    return total

重构后的代码

class OrderProcessor:
    """订单处理器，封装订单处理逻辑"""
    def __init__(self):
        self.logger = Logger()
    def process_order(self, order_data: dict) -> float:
        """处理订单"""
        self._validate_order(order_data)
        total = self._calculate_total(order_data['items'])
        total = self._apply_discount(total)
        self._log_result(total)
        return total
    def _validate_order(self, order_data: dict):
        """验证订单数据"""
        if not order_data.get('user_id'):
            raise ValueError("用户ID不能为空")
        items = order_data.get('items', [])
        if not items:
            raise ValueError("订单项不能为空")
    def _calculate_total(self, items: list) -> float:
        """计算订单总价"""
        return sum(item['price'] * item['quantity'] for item in items)
    def _apply_discount(self, total: float) -> float:
        """应用折扣策略"""
        discount_strategy = DiscountStrategy()
        return discount_strategy.apply_discount(total)
    def _log_result(self, total: float):
        """记录处理结果"""
        self.logger.info(f"订单处理完成，总价: {total}")
class DiscountStrategy:
    """折扣策略封装"""
    def apply_discount(self, total: float) -> float:
        """根据总价应用不同折扣"""
        if total > 1000:
            return total * 0.9
        elif total > 500:
            return total * 0.95
        return total
class Logger:
    """日志记录器封装"""
    @staticmethod
    def info(message: str):
        """记录信息日志"""
        print(f"[INFO] {message}")

数据验证封装

原始代码

# 散落的验证逻辑
def register_user(username, email, age, password):
    if len(username) < 3:
        return "用户名太短"
    if len(username) > 20:
        return "用户名太长"
    if '@' not in email:
        return "邮箱格式错误"
    if age < 18:
        return "年龄不符合要求"
    if len(password) < 8:
        return "密码太短"
    # ... 更多业务逻辑

重构后的代码

from dataclasses import dataclass
from typing import List, Optional
@dataclass
class ValidationResult:
    """验证结果封装"""
    is_valid: bool
    errors: List[str]
    @classmethod
    def success(cls):
        return cls(is_valid=True, errors=[])
    @classmethod
    def failure(cls, errors: List[str]):
        return cls(is_valid=False, errors=errors)
class UserValidator:
    """用户数据验证器"""
    def validate_registration(self, user_data: dict) -> ValidationResult:
        """验证用户注册数据"""
        errors = []
        errors.extend(self._validate_username(user_data.get('username', '')))
        errors.extend(self._validate_email(user_data.get('email', '')))
        errors.extend(self._validate_age(user_data.get('age', 0)))
        errors.extend(self._validate_password(user_data.get('password', '')))
        if errors:
            return ValidationResult.failure(errors)
        return ValidationResult.success()
    def _validate_username(self, username: str) -> List[str]:
        """验证用户名"""
        errors = []
        if len(username) < 3:
            errors.append("用户名至少需要3个字符")
        if len(username) > 20:
            errors.append("用户名不能超过20个字符")
        if not username.isalnum():
            errors.append("用户名只能包含字母和数字")
        return errors
    def _validate_email(self, email: str) -> List[str]:
        """验证邮箱"""
        errors = []
        if '@' not in email:
            errors.append("邮箱必须包含@符号")
        if '.' not in email.split('@')[-1]:
            errors.append("邮箱域名格式不正确")
        return errors
    def _validate_age(self, age: int) -> List[str]:
        """验证年龄"""
        errors = []
        if age < 18:
            errors.append("年龄必须大于18岁")
        if age > 120:
            errors.append("年龄似乎不太现实")
        return errors
    def _validate_password(self, password: str) -> List[str]:
        """验证密码"""
        errors = []
        if len(password) < 8:
            errors.append("密码至少需要8个字符")
        if not any(c.isupper() for c in password):
            errors.append("密码需要包含大写字母")
        if not any(c.islower() for c in password):
            errors.append("密码需要包含小写字母")
        if not any(c.isdigit() for c in password):
            errors.append("密码需要包含数字")
        return errors
# 使用示例
class UserRegistration:
    """用户注册处理器"""
    def __init__(self):
        self.validator = UserValidator()
    def register(self, user_data: dict) -> dict:
        """处理用户注册"""
        validation_result = self.validator.validate_registration(user_data)
        if not validation_result.is_valid:
            return {
                'success': False,
                'errors': validation_result.errors
            }
        # 实际的注册逻辑
        return {
            'success': True,
            'message': '注册成功'
        }

策略模式重构

原始代码

# 大量的条件判断
def calculate_shipping_cost(order, shipping_type):
    if shipping_type == 'standard':
        cost = order.weight * 0.5
        if order.weight > 10:
            cost += 5
    elif shipping_type == 'express':
        cost = order.weight * 1.0
        cost += 10
    elif shipping_type == 'overnight':
        cost = order.weight * 2.0
        cost += 20
        if order.urgent:
            cost += 15
    return cost

重构后的代码

from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class Order:
    """订单数据封装"""
    weight: float
    urgent: bool = False
class ShippingStrategy(ABC):
    """运费策略基类"""
    @abstractmethod
    def calculate(self, order: Order) -> float:
        """计算运费"""
        pass
class StandardShipping(ShippingStrategy):
    """标准配送策略"""
    def calculate(self, order: Order) -> float:
        cost = order.weight * 0.5
        if order.weight > 10:
            cost += 5
        return cost
class ExpressShipping(ShippingStrategy):
    """快递配送策略"""
    def calculate(self, order: Order) -> float:
        return order.weight * 1.0 + 10
class OvernightShipping(ShippingStrategy):
    """次日达配送策略"""
    def calculate(self, order: Order) -> float:
        cost = order.weight * 2.0 + 20
        if order.urgent:
            cost += 15
        return cost
class ShippingCalculator:
    """运费计算器"""
    def __init__(self):
        self._strategies = {
            'standard': StandardShipping(),
            'express': ExpressShipping(),
            'overnight': OvernightShipping()
        }
    def calculate(self, order: Order, shipping_type: str) -> float:
        """计算运费"""
        strategy = self._strategies.get(shipping_type)
        if not strategy:
            raise ValueError(f"不支持的配送类型: {shipping_type}")
        return strategy.calculate(order)
    def add_strategy(self, name: str, strategy: ShippingStrategy):
        """动态添加策略"""
        self._strategies[name] = strategy
# 使用示例
calculator = ShippingCalculator()
order = Order(weight=15, urgent=True)
cost = calculator.calculate(order, 'overnight')
print(f"运费: {cost}元")

数据管道封装

原始代码

# 混杂的数据处理逻辑
def process_data(data):
    # 清洗
    cleaned = []
    for item in data:
        if item is not None and item != '':
            if isinstance(item, str):
                item = item.strip()
            cleaned.append(item)
    # 转换
    transformed = []
    for item in cleaned:
        if isinstance(item, str):
            item = item.lower()
        transformed.append(item)
    # 过滤
    filtered = []
    for item in transformed:
        if isinstance(item, str) and len(item) > 3:
            filtered.append(item)
        elif isinstance(item, (int, float)) and item > 0:
            filtered.append(item)
    return filtered

重构后的代码

from typing import Any, List, Callable
from functools import reduce
class DataPipeline:
    """数据处理管道"""
    def __init__(self):
        self._stages: List[Callable] = []
    def add_stage(self, stage: Callable) -> 'DataPipeline':
        """添加处理阶段"""
        self._stages.append(stage)
        return self
    def process(self, data: Any) -> Any:
        """执行管道处理"""
        return reduce(lambda d, stage: stage(d), self._stages, data)
class DataProcessor:
    """数据处理器，提供各种处理函数"""
    @staticmethod
    def clean_data(data: List) -> List:
        """数据清洗"""
        return [
            item.strip() if isinstance(item, str) and item.strip() 
            else item 
            for item in data 
            if item is not None
        ]
    @staticmethod
    def transform_data(data: List) -> List:
        """数据转换"""
        return [
            item.lower() if isinstance(item, str) 
            else item 
            for item in data
        ]
    @staticmethod
    def filter_data(data: List) -> List:
        """数据过滤"""
        result = []
        for item in data:
            if isinstance(item, str) and len(item) > 3:
                result.append(item)
            elif isinstance(item, (int, float)) and item > 0:
                result.append(item)
        return result
    @staticmethod
    def deduplicate_data(data: List) -> List:
        """数据去重"""
        seen = set()
        result = []
        for item in data:
            key = item if isinstance(item, (str, int, float)) else str(item)
            if key not in seen:
                seen.add(key)
                result.append(item)
        return result
    @staticmethod
    def sort_data(data: List, reverse: bool = False) -> List:
        """数据排序"""
        return sorted(data, reverse=reverse)
# 使用示例
pipeline = DataPipeline()
processor = DataProcessor()
pipeline.add_stage(processor.clean_data)
pipeline.add_stage(processor.transform_data)
pipeline.add_stage(processor.filter_data)
pipeline.add_stage(processor.deduplicate_data)
pipeline.add_stage(processor.sort_data)
# 特定场景的自定义管道
custom_pipeline = DataPipeline()
custom_pipeline.add_stage(processor.clean_data)
custom_pipeline.add_stage(processor.filter_data)
# 处理数据
raw_data = ["  Hello  ", "WORLD", "", None, 123, "  ", "hello", -5, 10]
result = pipeline.process(raw_data)
print(f"处理结果: {result}")

重构工具建议

常用重构工具

IDE内置工具
- PyCharm/IntelliJ IDEA: Extract Method, Extract Class, Rename
- VS Code + Python扩展: Code Actions

专门的重构库

# 使用 rope 进行自动化重构
import rope.base.project
from rope.refactor.extract import ExtractMethod

project = rope.base.project.Project('.') resource = project.find_module('my_module') extracted = ExtractMethod(project, resource, start_offset, end_offset) changes = extracted.get_changes('new_method_name') project.do(changes)


3. **代码质量检查**
```bash
# 安装 pylint 和 autopep8
pip install pylint autopep8
# 检查代码质量
pylint my_module.py
# 自动格式化
autopep8 --in-place --aggressive my_module.py

单一职责原则: 每个类/函数只做一件事
开闭原则: 对扩展开放，对修改关闭
依赖倒置: 依赖抽象，不依赖具体实现
接口隔离: 最小化接口依赖
DRY原则: 不要重复自己

通过这些案例,你可以看到好的封装可以：