本文目录导读:
- 核心思想:从“写死”到“参数化”
- 面向对象设计:让复用更有层次
- 模块化设计:拆分与组合
- 配置分离:代码和数据分离
- 高阶技巧:装饰器与上下文管理器
- 实战项目模板:快速复用的架构
- 最佳实践清单
- 一个完整的复用案例(总结)
Python 案例的灵活复用,核心在于抽象、模块化和设计模式的合理运用,下面我帮你梳理几个最实用、最核心的复用技巧,从基础到进阶,让你写出的代码能“一次编写,多处运行”。
核心思想:从“写死”到“参数化”
最直接的复用方式,就是让代码不要硬编码具体数值,而是通过参数来控制行为。
案例:统计文件行数
不灵活的方式(硬编码):
# 只能统计这一个文件
with open('data.txt', 'r') as f:
lines = f.readlines()
print(f"总行数:{len(lines)}")
灵活可复用的方式(参数化):
def count_lines(filename):
"""通用函数:统计任何文本文件的行数"""
with open(filename, 'r') as f:
return len(f.readlines())
# 复用时只需换文件名
print(count_lines('data.txt'))
print(count_lines('log.txt'))
print(count_lines('config.ini'))
进一步扩展(带可选参数):
def count_lines(filename, ignore_empty=False, encoding='utf-8'):
"""更通用的行数统计函数"""
with open(filename, 'r', encoding=encoding) as f:
lines = f.readlines()
if ignore_empty:
lines = [l for l in lines if l.strip()]
return len(lines)
# 不同场景不同用法
count_lines('data.txt') # 默认统计所有行
count_lines('data.txt', ignore_empty=True) # 忽略空行
count_lines('data.txt', encoding='gbk') # 指定编码
核心原则: 只要可能就让具体的数值、路径、配置变成函数参数。
面向对象设计:让复用更有层次
当案例逻辑复杂时,用类来封装状态和行为,比函数更易扩展和维护。
案例:简单的数据处理流程
基础类设计:
class DataProcessor:
"""数据处理基类,定义骨架,子类实现细节"""
def __init__(self, source):
self.source = source
self.data = []
def load(self):
"""加载数据,子类可重写"""
raise NotImplementedError
def clean(self):
"""数据清洗,默认实现"""
self.data = [d for d in self.data if d is not None]
return self
def process(self):
"""核心处理,留给子类实现"""
raise NotImplementedError
def run(self):
"""执行完整流程:加载 -> 清洗 -> 处理"""
self.load()
self.clean()
return self.process()
复用方式1:继承并重写
class CSVProcessor(DataProcessor):
def load(self):
import csv
with open(self.source, 'r') as f:
reader = csv.DictReader(f)
self.data = [row for row in reader]
def process(self):
# 专门处理CSV的逻辑
return [d for d in self.data if float(d['score']) > 80]
class JSONProcessor(DataProcessor):
def load(self):
import json
with open(self.source, 'r') as f:
self.data = json.load(f)
def process(self):
# 专门处理JSON的逻辑
return {'total': len(self.data), 'items': self.data[:10]}
# 使用
csv_result = CSVProcessor('scores.csv').run()
json_result = JSONProcessor('config.json').run()
复用方式2:组合(依赖注入)
class FlexibleProcessor:
"""通过外部注入不同策略,实现横向复用"""
def __init__(self, loader, cleaner, processor):
self.loader = loader # 可以是任何实现了__call__的对象
self.cleaner = cleaner
self.processor = processor
def run(self, source):
data = self.loader(source)
data = self.cleaner(data)
return self.processor(data)
# 定义不同策略(甚至可以是lambda)
csv_loader = lambda src: [{'name':'Alice', 'score':90}]
json_loader = lambda src: {'users': []}
processor = FlexibleProcessor(csv_loader, lambda d: d, lambda d: d)
result = processor.run('file.csv')
模块化设计:拆分与组合
把功能拆成多个小模块,每个模块做好一件事。
案例:数据分析结果导出
按功能模块拆分:
# analysis/analyzers.py —— 分析逻辑
def calculate_statistics(data):
return {
'mean': sum(data)/len(data),
'max': max(data),
'min': min(data)
}
# analysis/exporters.py —— 导出逻辑
def to_json(data, filepath):
import json
with open(filepath, 'w') as f:
json.dump(data, f)
def to_csv(data, filepath):
import csv
with open(filepath, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(data.keys())
writer.writerow(data.values())
# analysis/formatters.py —— 格式化逻辑
def format_for_report(stats):
return f"均值:{stats['mean']:.2f}\n最大值:{stats['max']}"
# main.py —— 组合使用
from analysis.analyzers import calculate_statistics
from analysis.exporters import to_json, to_csv
data = [85, 92, 78, 90, 88]
stats = calculate_statistics(data)
to_json(stats, 'stats.json')
to_csv(stats, 'stats.csv')
关键好处:
- 任何一个步骤都可以被独立复用(单独在别的项目里import)
- 可以轻松替换某个模块(比如换了导出格式,只需更换exporter)
- 便于并行开发和测试
配置分离:代码和数据分离
把可变部分写成配置文件,让案例在完全不改代码的情况下复用。
案例:网络爬虫
# config/settings.py —— 配置集中管理
SCRAPER_CONFIG = {
'headers': {
'User-Agent': 'Mozilla/5.0',
'Accept': 'text/html'
},
'timeout': 10,
'retry_times': 3,
'delay': 1.0, # 请求间隔秒数
'output_format': 'json',
'max_pages': 50
}
# crawler.py —— 使用配置
from config.settings import SCRAPER_CONFIG
class WebCrawler:
def __init__(self, config=None):
self.config = config or SCRAPER_CONFIG # 允许传入自定义配置
self.session = requests.Session()
self.session.headers.update(self.config['headers'])
def crawl(self, urls):
results = []
for url in urls:
resp = self._fetch(url)
results.append(resp)
time.sleep(self.config['delay'])
return results
# 不同场景复用:
# 场景1:使用默认配置
crawler = WebCrawler()
data = crawler.crawl(['https://example.com'])
# 场景2:自定义配置
custom_config = {**SCRAPER_CONFIG, 'delay': 0.5, 'max_pages': 100}
fast_crawler = WebCrawler(config=custom_config)
更优方案:使用配置文件(YAML/TOML/JSON)
# config.yaml database: host: localhost port: 3306 name: mydb api: base_url: "https://api.example.com" version: v2
import yaml
with open('config.yaml', 'r') as f:
config = yaml.safe_load(f)
# 现在只需换config文件,整个程序就能切换环境/场景
connect_to_db(config['database'])
高阶技巧:装饰器与上下文管理器
这些特性可以让你在不修改原有函数/类代码的情况下,为其添加新功能。
案例:给函数添加日志/计时/缓存
import time
import functools
# 复用度最高的装饰器:计时
def timer(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
cost = time.time() - start
print(f"[{func.__name__}] 耗时:{cost:.3f}秒")
return result
return wrapper
# 可配置的缓存装饰器
def cache_result(maxsize=128):
def decorator(func):
cache = {}
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
if key not in cache:
cache[key] = func(*args, **kwargs)
if len(cache) > maxsize:
cache.pop(next(iter(cache))) # 移除最早的
return cache[key]
return wrapper
return decorator
# 实现复用:任何函数都能使用
@timer
def fetch_data(api_url):
time.sleep(1) # 模拟网络请求
return {'status': 'ok'}
@cache_result(maxsize=10)
def expensive_calc(n):
time.sleep(0.5) # 模拟耗时计算
return n ** 100
# 计时功能在调试/监控时随处可用
@timer
def save_to_db(data):
pass # 数据库操作
上下文管理器:让资源管理复用
from contextlib import contextmanager
@contextmanager
def timer_block(name="block"):
"""可以复用的计时上下文,用于代码块级别的计时"""
print(f"开始 {name}")
start = time.time()
try:
yield
finally:
cost = time.time() - start
print(f"结束 {name},耗时:{cost:.3f}秒")
# 任何地方都可以这样使用
with timer_block("数据处理"):
data = load_data()
process(data)
save_data(data)
实战项目模板:快速复用的架构
当你需要频繁创建类似项目时,用模板、工厂模式结合参数化。
# project_factory.py —— 项目生成器
import os
import shutil
class ProjectGenerator:
"""根据模板快速生成Python项目骨架"""
def __init__(self, template_dir='templates/base'):
self.template_dir = template_dir
def generate(self, project_name, author, output_dir='.'):
target = os.path.join(output_dir, project_name)
if os.path.exists(target):
raise FileExistsError(f"{target} 已存在")
# 复制模板
shutil.copytree(self.template_dir, target)
# 替换模板中的占位符
for dirpath, _, filenames in os.walk(target):
for filename in filenames:
if filename.endswith(('.py', '.md', '.txt')):
self._replace_placeholders(
os.path.join(dirpath, filename),
project_name=project_name,
author=author
)
print(f"项目 {project_name} 创建成功!")
def _replace_placeholders(self, filepath, **kwargs):
with open(filepath, 'r') as f:
content = f.read()
for key, value in kwargs.items():
content = content.replace(f"{{{{{key}}}}}", value)
with open(filepath, 'w') as f:
f.write(content)
# 使用:秒级创建新项目
gen = ProjectGenerator()
gen.generate('my_analysis_tool', author='张三')
gen.generate('my_web_crawler', author='李四', output_dir='projects/')
最佳实践清单
✅ 核心要做的
- 写函数,不写脚本:一切逻辑封装成函数或类
- 每个函数只做一件事:单一职责原则,便于组合
- 参数化所有可变值:路径、配置、阈值、格式等
- 定义清晰的接口:使用类型注解(Type Hints)让参数和返回值明确
- 提供默认值:让最常用场景不需要传参
- 文档字符串(Docstring):写明用途、参数、返回值、示例
✅ 常见复用模式速查
| 场景 | 最佳复用方式 | 代码示例 |
|---|---|---|
| 重复的计算逻辑 | 函数 + 参数 | calc(x, method='mean') |
| 不同格式的文件处理 | 类继承 | class CSVHandler(FileHandler) |
| 时间、日志、性能监控 | 装饰器 | @timer |
| 资源管理(锁、连接) | 上下文管理器 | with db_session(): |
| 依赖外部服务(API/DB) | 配置 + 工厂函数 | create_api_client(config) |
| 多处使用相同数据集 | 缓存 | @lru_cache |
| 业务逻辑类似但细节不同 | 策略模式(函数作为参数) | process(data, strategy=moving_average) |
一个完整的复用案例(
这里展示一个模拟邮件发送系统的设计,展示了 配置分离 + 接口抽象 + 多种实现复用:
from abc import ABC, abstractmethod
import smtplib
import logging
# ===== 1. 抽象接口(基类)======
class MailSender(ABC):
@abstractmethod
def send(self, to, subject, body):
pass
# ===== 2. 多种实现,可随意切换 =====
class SMTPSender(MailSender):
def __init__(self, host, port, user, password, use_tls=True):
self.host = host
self.port = port
self.user = user
self.password = password
self.use_tls = use_tls
def send(self, to, subject, body):
msg = f"Subject: {subject}\n\n{body}"
with smtplib.SMTP(self.host, self.port) as s:
if self.use_tls:
s.starttls()
s.login(self.user, self.password)
s.sendmail(self.user, to, msg)
logging.info(f"邮件已发送给 {to}")
class PrintSender(MailSender):
"""调试用,只打印不实际发送"""
def send(self, to, subject, body):
logging.info(f"[模拟] 发送给 {to}: {subject}")
# ===== 3. 配置驱动,零改动切换 =====
import json
def create_mail_sender_from_config(config_path):
"""从配置文件读取配置,创建对应的sender实例"""
with open(config_path, 'r') as f:
config = json.load(f)
sender_type = config.get('type', 'smtp')
if sender_type == 'smtp':
return SMTPSender(**config['smtp'])
elif sender_type == 'print':
return PrintSender()
else:
raise ValueError(f"未知类型: {sender_type}")
# ===== 4. 使用:只需换配置文件 =====
# config_dev.json → 使用打印模式,无实际发送
# config_prod.json → 使用真实SMTP
sender = create_mail_sender_from_config('config_prod.json')
sender.send('user@example.com', 'Hello', 'This is a test.')
这种设计下:
- 如果要增加新的发送方式(比如API发送),只需新增一个
APISender类,不需要改其他任何代码 - 配置放在外部,业务逻辑无需调整
- 调试和生产环境一键切换
总结一句话:Python案例灵活复用的关键就是 “把变化的部分抽象出来,通过参数、配置、策略、模板等方式注入”,让不变的核心逻辑成为可复用的基础设施。
标签: 参数化配置