Python案例如何划分函数

wen python案例 2026-06-05 325

本文目录导读：

Python案例如何划分函数

核心划分原则
实战案例：商品管理系统
函数划分检查清单
实际项目中的函数划分示例

在Python中合理划分函数是写出高质量代码的关键,我来分享几个实用的划分原则和具体案例。

核心划分原则

单一职责原则

每个函数只做一件事,做好一件事。

# ❌ 不好的设计：一个函数做太多事
def process_user_data(user_id):
    # 1. 从数据库获取用户
    user = db.query(f"SELECT * FROM users WHERE id={user_id}")
    # 2. 验证数据
    if not user.get('email'):
        raise ValueError("邮箱不能为空")
    # 3. 发送邮件
    send_email(user['email'], "欢迎注册")
    # 4. 更新统计
    update_stats()
    return user
# ✅ 好的设计：拆分成多个单一职责函数
def get_user(user_id):
    """只负责获取用户数据"""
    return db.query(f"SELECT * FROM users WHERE id={user_id}")
def validate_user(user):
    """只负责验证用户数据"""
    if not user.get('email'):
        raise ValueError("邮箱不能为空")
def send_welcome_email(user):
    """只负责发送欢迎邮件"""
    send_email(user['email'], "欢迎注册")
def process_user_data(user_id):
    """协调者：只负责调用其他函数"""
    user = get_user(user_id)
    validate_user(user)
    send_welcome_email(user)
    update_stats()
    return user

基于抽象级别划分

将不同抽象级别的操作分开。

# 底层函数：具体实现细节
def read_file_content(filepath):
    """读取文件内容（底层操作）"""
    with open(filepath, 'r', encoding='utf-8') as f:
        return f.read()
def parse_json_string(json_str):
    """解析JSON字符串（底层操作）"""
    import json
    return json.loads(json_str)
# 中层函数：组合底层操作
def load_config_from_file(filepath):
    """从文件加载配置（中层操作）"""
    content = read_file_content(filepath)
    return parse_json_string(content)
# 高层函数：业务逻辑
def initialize_application(config_path):
    """初始化应用（高层操作）"""
    config = load_config_from_file(config_path)
    setup_database(config['database'])
    setup_logging(config['logging'])
    return config

数据流向划分

根据数据如何输入、处理和输出进行划分。

def get_raw_data(source):
    """数据输入：获取原始数据"""
    return source.fetch_all()
def transform_data(raw_data):
    """数据处理：转换数据格式"""
    processed = []
    for item in raw_data:
        processed.append({
            'id': item[0],
            'name': item[1].strip(),
            'price': float(item[2]) * 1.1  # 加10%税
        })
    return processed
def validate_data(transformed_data):
    """数据验证：确保数据质量"""
    valid_data = []
    for item in transformed_data:
        if item['price'] > 0 and item['name']:
            valid_data.append(item)
    return valid_data
def save_data(valid_data, output_path):
    """数据输出：保存处理后的数据"""
    import json
    with open(output_path, 'w') as f:
        json.dump(valid_data, f)
# 主流程函数
def process_data_pipeline(source, output_path):
    """数据处理流水线"""
    raw_data = get_raw_data(source)
    transformed_data = transform_data(raw_data)
    valid_data = validate_data(transformed_data)
    save_data(valid_data, output_path)

实战案例：商品管理系统

class ProductManager:
    """商品管理系统"""
    # ========== 数据持久化层 ==========
    def _save_to_database(self, products):
        """保存商品到数据库"""
        # 数据库操作实现
        pass
    def _load_from_database(self):
        """从数据库加载商品"""
        # 数据库操作实现
        pass
    # ========== 业务逻辑层 ==========
    def add_product(self, name, price, quantity):
        """添加新商品"""
        product = self._create_product(name, price, quantity)
        self._validate_product(product)
        self._save_to_database([product])
        return product
    def update_price(self, product_id, new_price):
        """更新商品价格"""
        product = self._get_product_by_id(product_id)
        self._validate_price(new_price)
        product['price'] = new_price
        self._update_product(product)
    def check_inventory(self, product_id):
        """检查库存"""
        product = self._get_product_by_id(product_id)
        return product['quantity']
    def reduce_quantity(self, product_id, amount):
        """减少库存数量"""
        product = self._get_product_by_id(product_id)
        if product['quantity'] < amount:
            raise ValueError(f"库存不足，当前库存: {product['quantity']}")
        product['quantity'] -= amount
        self._update_product(product)
    # ========== 辅助方法 ==========
    def _create_product(self, name, price, quantity):
        """创建商品对象"""
        import uuid
        return {
            'id': str(uuid.uuid4()),
            'name': name,
            'price': price,
            'quantity': quantity
        }
    def _validate_product(self, product):
        """验证商品数据"""
        if not product['name']:
            raise ValueError("商品名称不能为空")
        self._validate_price(product['price'])
        if product['quantity'] < 0:
            raise ValueError("库存数量不能为负数")
    def _validate_price(self, price):
        """验证价格"""
        if price <= 0:
            raise ValueError("价格必须大于0")
    def _get_product_by_id(self, product_id):
        """根据ID获取商品"""
        products = self._load_from_database()
        for product in products:
            if product['id'] == product_id:
                return product
        raise ValueError(f"商品ID {product_id} 不存在")
    def _update_product(self, product):
        """更新商品信息到数据库"""
        # 更新数据库中的商品
        pass
# 使用示例
manager = ProductManager()
product = manager.add_product("笔记本电脑", 5999.99, 10)
manager.reduce_quantity(product['id'], 1)

函数划分检查清单

在决定是否拆分函数时,可以问自己这些问题：

# 应该拆分的信号
def should_split_function(func):
    """检查函数是否需要拆分"""
    signals = []
    # 1. 函数太长（超过20-30行）
    signals.append(len(func.split('\n')) > 30)
    # 2. 缩进层次太多（超过3层）
    # （需要实际检查缩进）
    # 3. 函数名称中出现"and"
    signals.append('and' in func.__name__.lower())
    # 4. 有重复代码块
    # （需要实际检查）
    # 5. 需要大量注释解释
    # （需要实际检查）
    # 6. 有多个可独立测试的逻辑单元
    # （需要实际检查）
    return any(signals)

实际项目中的函数划分示例

# 一个完整的用户注册流程
class UserRegistration:
    """用户注册服务"""
    # 公有接口
    def register(self, username, password, email):
        """注册新用户"""
        user_data = self._create_user_data(username, password, email)
        self._validate_registration_data(user_data)
        self._check_duplicate_user(username, email)
        self._generate_user_id(user_data)
        self._hash_password(user_data)
        self._save_user(user_data)
        self._send_welcome_email(user_data)
        return user_data
    # 私有实现方法
    def _create_user_data(self, username, password, email):
        """创建用户数据对象"""
        return {
            'username': username.strip(),
            'password': password,
            'email': email.lower().strip(),
            'created_at': '2024-01-01',
            'is_active': True
        }
    def _validate_registration_data(self, user_data):
        """验证注册数据"""
        errors = []
        if len(user_data['username']) < 3:
            errors.append("用户名至少3个字符")
        if len(user_data['password']) < 8:
            errors.append("密码至少8个字符")
        if '@' not in user_data['email']:
            errors.append("邮箱格式不正确")
        if errors:
            raise ValueError("; ".join(errors))
    def _check_duplicate_user(self, username, email):
        """检查用户是否已存在"""
        # 查询数据库
        pass
    def _generate_user_id(self, user_data):
        """生成用户ID"""
        import uuid
        user_data['user_id'] = str(uuid.uuid4())
    def _hash_password(self, user_data):
        """密码哈希"""
        import hashlib
        plain_password = user_data['password']
        user_data['password_hash'] = hashlib.sha256(
            plain_password.encode()
        ).hexdigest()
        del user_data['password']  # 删除明文密码
    def _save_user(self, user_data):
        """保存用户到数据库"""
        # 数据库操作
        pass
    def _send_welcome_email(self, user_data):
        """发送欢迎邮件"""
        # 邮件发送逻辑
        pass

划分函数的核心原则：