Python案例怎样复用函数

wen python案例 2026-06-04 43

Python函数复用实战：从基础案例到高效代码架构

📚 目录导读

为什么函数复用是Python开发的基石
函数复用的核心技巧：参数化与模块化
实战案例1：数据清洗函数的多场景复用
实战案例2：API调用函数的通用封装
实战案例3：装饰器实现日志与性能监控
函数复用的高阶玩法：闭包与生成器
常见错误与最佳实践Q&A
从“写函数”到“设计函数”

为什么函数复用是Python开发的基石

在Python开发中，函数复用不仅是减少重复代码的手段，更是构建可维护、可扩展系统的核心能力，许多新手容易陷入“复制粘贴-修改”的循环，导致代码冗余、调试困难，当你需要处理多个不同格式的日期字符串时，若每次都写一套strptime逻辑,代码将快速膨胀。

Python案例怎样复用函数

一个核心原则：函数复用是对“单一职责原则”的实践——每个函数只做一件事,但通过参数和返回值适应不同场景。

函数复用的核心技巧：参数化与模块化

1 参数化：让函数适配不同输入

# 非复用版：硬编码路径
def load_csv_v1():
    df = pd.read_csv('./data/sales.csv')
    return df
# 复用版：参数化路径与分隔符
def load_csv(file_path, delimiter='，'):
    df = pd.read_csv(file_path, sep=delimiter)
    return df

2 模块化：将函数组织成独立模块

创建utils/data_processor.py文件,将通用函数集中管理：

# utils/data_processor.py
def clean_whitespace(text):
    return ' '.join(text.split())
def validate_email(email):
    import re
    pattern = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
    return re.match(pattern, email) is not None

在业务代码中from utils.data_processor import clean_whitespace即可复用。

实战案例1：数据清洗函数的多场景复用

场景：清洗用户输入的手机号，但不同来源格式不同（如“138 1234 5678”、“+86-13812345678”）。

1 一次性清洗函数

def clean_phone_raw(phone):
    phone = phone.replace(' ', '').replace('-', '')
    phone = phone.replace('+86', '')
    if len(phone) != 11:
        return None
    return phone

2 可复用的参数化版本

def clean_phone(phone, remove_prefix='+86', valid_length=11):
    import re
    # 去除所有非数字字符
    digits = re.sub(r'\D', '', phone)
    # 去除前缀
    if digits.startswith(remove_prefix):
        digits = digits[len(remove_prefix):]
    return digits if len(digits) == valid_length else None

复用场景：当需要清洗不同国家手机号时，只需传remove_prefix='+1'、valid_length=10,无需重写逻辑。

实战案例2：API调用函数的通用封装

1 非复用代码（每次调用都要写try-except）

import requests
response = requests.get('https://api.example.com/user', headers={'Auth': 'xxx'})
if response.status_code == 200:
    data = response.json()
else:
    raise Exception(f'API error: {response.status_code}')

2 通用API调用函数

def call_api(url, method='GET', params=None, headers=None, timeout=10):
    import requests
    try:
        response = requests.request(method, url, params=params, headers=headers, timeout=timeout)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f'API call failed: {url}, error: {str(e)}')
        return None
# 复用示例
user_data = call_api('https://api.example.com/user', headers={'Auth': 'xxx'})
product_data = call_api('https://api.example.com/products', params={'category': 'electronics'})

3 扩展到重试机制

利用装饰器实现自动重试：

import time
from functools import wraps
def retry(max_retries=3, delay=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                result = func(*args, **kwargs)
                if result is not None:
                    return result
                time.sleep(delay)
            return None
        return wrapper
    return decorator
@retry(max_retries=5, delay=1)
def call_api_with_retry(url, **kwargs):
    return call_api(url, **kwargs)

实战案例3：装饰器实现日志与性能监控

1 日志记录装饰器

import logging
logging.basicConfig(level=logging.INFO)
def log_function_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f'Calling {func.__name__} with args={args}, kwargs={kwargs}')
        result = func(*args, **kwargs)
        logging.info(f'{func.__name__} returned: {result}')
        return result
    return wrapper
@log_function_call
def calculate_discount(price, rate):
    return price * rate

2 性能计时装饰器

import time
def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f'{func.__name__} took {end - start:.4f} seconds')
        return result
    return wrapper
@timer
def compute_large_dataset():
    # 模拟耗时操作
    time.sleep(2)
    return 'done'

复用价值：一个@timer装饰器可应用于任何需要性能分析的函数,无需侵入原有业务逻辑。

函数复用的高阶玩法：闭包与生成器

1 闭包实现状态保持

def make_counter(start=0):
    count = [start]  # 使用列表实现可变变量
    def counter():
        count[0] += 1
        return count[0]
    return counter
counter_a = make_counter(10)
print(counter_a())  # 11
print(counter_a())  # 12
counter_b = make_counter(100)  # 独立状态
print(counter_b())  # 101

2 生成器实现流式复用

def read_large_file(file_path, chunk_size=1024):
    with open(file_path, 'r') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk
# 复用：无论处理什么大文件，只需传入不同路径
for chunk in read_large_file('huge_log.txt'):
    process_chunk(chunk)