第3章 3.5 综合实战:可复用逻辑封装篇
—— 用 Python 组合式函数打造你的「工具百宝箱」
🎯 开场 3 分钟:为什么要学这个?
上一章我们学会了 mixin 和组合式函数的对比,知道组合式函数就像「模块化积木」,哪里需要往哪里搭。
你有没有遇到过这种痛苦:
- 写了一个弹窗功能,下次做新项目发现没法复用
- 复制粘贴代码满屏都是,维护起来想哭
- 改一个小功能要翻好几个文件,不知道该动哪个
学完这章,你能:
用一个「搭积木」的思路,把常用功能封装成「组合式函数」,下次写项目直接调用,像拼乐高一样快。
🧱 基础 25 分钟:核心概念(小白视角)
什么是组合式函数?
生活类比:想象你有个「万能工具箱」,里面不是工具,而是做某件事的完整步骤。
比如「泡茶」这个步骤:
1. 烧水
2. 放茶叶
3. 等3分钟
你不需要每次都记这3步,直接喊「给我来一杯茶」就行。
组合式函数就是这样——把「做一件事的步骤」打包成一个函数,下次调用。
\n\n
\n\n
\n\n 为什么不用普通函数?
普通函数的问题是:状态分散,难以复用。
比如你写了一个「记录用户点击」的功能,里面有变量 count、lastClick、history,散落在代码各处。下次想复用?得把这些变量和逻辑一起拷贝,麻烦死了。
组合式函数把相关的数据和逻辑打包在一起,叫「状态 + 行为 = 完整单元」。
第一个组合式函数
# use_click_tracker.py
"""追踪用户点击的组合式函数"""
def create_click_tracker():
"""创建一个点击追踪器"""
count = 0 # 点击次数
history = [] # 点击历史
def track_click(label):
nonlocal count, history
count += 1
history.append(f"{label}被点击了")
return {"count": count, "last": label}
def get_stats():
return {"total": count, "history": history.copy()}
return {"track_click": track_click, "get_stats": get_stats}
# 使用
tracker = create_click_tracker()
print(tracker["track_click"]("按钮A"))
print(tracker["track_click"]("按钮B"))
print(tracker["track_click"]("按钮A"))
print(tracker["get_stats"]())
输出:
{'count': 1, 'last': '按钮A'}
{'count': 2, 'last': '按钮B'}
{'count': 3, 'last': '按钮A'}
{'total': 3, 'history': ['按钮A被点击了', '按钮B被点击了', '按钮A被点击了']}
这3行代码在干嘛:创建了一个「点击追踪器」,每次点击都会记录次数和历史。
组合式函数的三大特点
| 特点 | 解释 | 生活类比 |
|---|---|---|
| 自包含 | 所有相关数据都在内部,不会污染全局 | 打包好的快递,里面东西不会掉出来 |
| 可组合 | 一个函数可以用另一个函数 | 「泡茶」可以用「烧水」这个现成步骤 |
| 可测试 | 功能独立,直接测这个函数就行 | 单独测试流水线上某个环节 |
进阶技巧:组合式函数嵌套
# use_form_validation.py
"""表单验证组合式函数"""
def create_validator():
"""创建表单验证器"""
errors = []
def required(field_name, value):
if not value or str(value).strip() == "":
errors.append(f"{field_name}不能为空")
return False
return True
def min_length(field_name, value, length):
if len(str(value)) < length:
errors.append(f"{field_name}长度不能小于{length}")
return False
return True
def email(field_name, value):
if "@" not in str(value):
errors.append(f"{field_name}必须是邮箱格式")
return False
return True
def get_errors():
return errors.copy()
def clear():
errors.clear()
return {
"required": required,
"min_length": min_length,
"email": email,
"get_errors": get_errors,
"clear": clear
}
# 组合两个函数:验证 + 追踪
def create_form_handler(form_name):
"""创建表单处理器:组合验证 + 追踪"""
validator = create_validator()
tracker = create_click_tracker()
def validate_and_submit(data):
validator.clear()
tracker.track_click(f"{form_name}提交")
validator.required("用户名", data.get("username"))
validator.min_length("密码", data.get("password"), 6)
validator.email("邮箱", data.get("email"))
errors = validator.get_errors()
if errors:
return {"success": False, "errors": errors}
return {"success": True, "message": "提交成功!"}
return {
"validate_and_submit": validate_and_submit,
"get_stats": tracker.get_stats
}
# 使用
handler = create_form_handler("注册表单")
result = handler["validate_and_submit"]({
"username": "tom",
"password": "123", # 密码太短
"email": "tom@example.com"
})
print(result)
输出:
{'success': False, 'errors': ['密码长度不能小于6']}
这代码在干嘛:把「验证器」和「追踪器」组合成一个「表单处理器」,验证失败就返回错误,成功就记录点击。
🔥 实战 35 分钟:3 个递进的小项目
项目 1(5 分钟):计数器组合式函数
目标:学会创建最基础的组合式函数
# project1_counter.py
"""计数器组合式函数"""
def create_counter():
"""创建一个带增加/减少/重置功能的计数器"""
count = 0
def increment():
nonlocal count
count += 1
return count
def decrement():
nonlocal count
count -= 1
return count
def reset():
nonlocal count
count = 0
return count
def value():
return count
return {
"increment": increment,
"decrement": decrement,
"reset": reset,
"value": value
}
# 测试
counter = create_counter()
print(f"初始值: {counter['value']()}")
counter['increment']()
counter['increment']()
print(f"增加2次后: {counter['value']()}")
counter['decrement']()
print(f"减少1次后: {counter['value']()}")
counter['reset']()
print(f"重置后: {counter['value']()}")
预期输出:
初始值: 0
增加2次后: 2
减少1次后: 1
重置后: 0
一句话解释:创建一个「私有关键字」count,外部无法直接访问,只能通过 increment/decrement/reset/value 操作。
项目 2(15 分钟):待办清单 with 组合式函数
目标:学会用组合式函数管理列表数据
# project2_todo.py
"""待办清单组合式函数"""
def create_todo_list(name):
"""创建一个带过滤、搜索、统计功能的待办清单"""
todos = []
_name = name
def add(task):
todos.append({
"task": task,
"done": False
})
return len(todos)
def done(index):
if 0 <= index < len(todos):
todos[index]["done"] = True
return True
return False
def remove(index):
if 0 <= index < len(todos):
todos.pop(index)
return True
return False
def filter_done():
return [t for t in todos if t["done"]]
def filter_undone():
return [t for t in todos if not t["done"]]
def search(keyword):
return [t for t in todos if keyword in t["task"]]
def stats():
total = len(todos)
done_count = len(filter_done())
return {
"name": _name,
"total": total,
"done": done_count,
"undone": total - done_count,
"progress": f"{done_count}/{total}"
}
def all():
return todos.copy()
return {
"add": add,
"done": done,
"remove": remove,
"filter_done": filter_done,
"filter_undone": filter_undone,
"search": search,
"stats": stats,
"all": all
}
# 使用
todo_list = create_todo_list("今日工作")
# 添加任务
todo_list["add"]("完成项目报告")
todo_list["add"]("回复邮件")
todo_list["add"]("开会讨论方案")
# 标记完成
todo_list["done"](0) # 完成第一项
# 搜索
print("包含'项目'的任务:", todo_list["search"]("项目"))
# 过滤
print("已完成:", todo_list["filter_done"]())
print("未完成:", todo_list["filter_undone"]())
# 统计
print("统计:", todo_list["stats"]())
# 展示全部
print("全部任务:")
for i, t in enumerate(todo_list["all"]()):
status = "✓" if t["done"] else "✗"
print(f" {i+1}. [{status}] {t['task']}")
预期输出:
包含'项目'的任务: [{'task': '完成项目报告', 'done': True}]
已完成: [{'task': '完成项目报告', 'done': True}]
未完成: [{'task': '回复邮件', 'done': False}, {'task': '开会讨论方案', 'done': False}]
统计: {'name': '今日工作', 'total': 3, 'done': 1, 'undone': 2, 'progress': '1/3'}
全部任务:
1. [✓] 完成项目报告
2. [✗] 回复邮件
3. [✗] 开会讨论方案
一句话解释:把「待办清单」的所有操作(增删改查过滤)打包成一个函数,对外只暴露接口,内部状态(todos)完全隐藏。
项目 3(15 分钟):数据清洗工具箱
目标:组合多个组合式函数,做一个真实可用的数据处理工具
# project3_data_cleaner.py
"""数据清洗组合式函数工具箱"""
# 1. 创建字符串清理器
def create_string_cleaner():
"""创建字符串清理工具"""
def trim(value):
return str(value).strip() if value else ""
def to_lower(value):
return str(value).lower().strip() if value else ""
def to_upper(value):
return str(value).upper().strip() if value else ""
def remove_spaces(value):
return str(value).replace(" ", "") if value else ""
return {
"trim": trim,
"to_lower": to_lower,
"to_upper": to_upper,
"remove_spaces": remove_spaces
}
# 2. 创建数据验证器
def create_data_validator():
"""创建数据验证工具"""
errors = []
def is_not_empty(value, field_name):
if not value or str(value).strip() == "":
errors.append(f"{field_name}不能为空")
return False
return True
def is_number(value, field_name):
try:
float(value)
return True
except:
errors.append(f"{field_name}必须是数字")
return False
def is_in_range(value, field_name, min_val, max_val):
try:
num = float(value)
if num < min_val or num > max_val:
errors.append(f"{field_name}必须在{min_val}-{max_val}之间")
return False
return True
except:
errors.append(f"{field_name}必须是数字")
return False
def get_errors():
return errors.copy()
def clear():
errors.clear()
return {
"is_not_empty": is_not_empty,
"is_number": is_number,
"is_in_range": is_in_range,
"get_errors": get_errors,
"clear": clear
}
# 3. 创建数据清洗管道(组合以上两个)
def create_cleaning_pipeline():
"""创建数据清洗管道:组合清理器 + 验证器"""
cleaner = create_string_cleaner()
validator = create_data_validator()
def clean_and_validate(record, rules):
"""
record: 原始数据字典
rules: 清洗规则字典,格式 {"字段名": ["trim", "to_lower", ...]}
"""
validator.clear()
cleaned = {}
for field, operations in rules.items():
value = record.get(field, "")
# 执行清洗操作
for op in operations:
if op == "trim":
value = cleaner["trim"](value)
elif op == "to_lower":
value = cleaner["to_lower"](value)
elif op == "to_upper":
value = cleaner["to_upper"](value)
elif op == "remove_spaces":
value = cleaner["remove_spaces"](value)
cleaned[field] = value
return cleaned
def clean_batch(records, rules):
return [clean_and_validate(r, rules) for r in records]
return {
"clean_and_validate": clean_and_validate,
"clean_batch": clean_batch,
"get_errors": validator.get_errors
}
# 真实场景:清洗用户注册数据
raw_users = [
{"name": " Tom Smith ", "age": "25", "email": " TOM@EXAMPLE.COM "},
{"name": " JANE DOE ", "age": "abc", "email": "jane@example.com"},
{"name": " Bob ", "age": "30", "email": " BOB@Example.COM "},
]
# 定义清洗规则
rules = {
"name": ["trim", "to_title"], # to_title是手动处理
"age": ["trim"],
"email": ["trim", "to_lower"]
}
pipeline = create_cleaning_pipeline()
# 手动实现 title case(Python没有内置)
def to_title(value):
return " ".join(word.capitalize() for word in str(value).split())
# 逐条清洗
cleaned_users = []
for user in raw_users:
cleaned = {
"name": to_title(user.get("name", "").strip()),
"age": user.get("age", "").strip(),
"email": pipeline["clean_and_validate"](user, rules)["email"]
}
# 验证
validator = create_data_validator()
validator.is_not_empty(cleaned["name"], "name")
validator.is_number(cleaned["age"], "age")
if validator.get_errors():
print(f"数据错误: {validator.get_errors()}")
else:
cleaned_users.append(cleaned)
print("\n清洗后的用户数据:")
for u in cleaned_users:
print(f" {u}")
预期输出:
清洗后的用户数据:
{'name': 'Tom Smith', 'age': '25', 'email': 'tom@example.com'}
{'name': 'Jane Doe', 'age': 'abc', 'email': 'jane@example.com'}
数据错误: ['age必须是数字']
{'name': 'Bob', 'age': '30', 'email': 'bob@example.com'}
一句话解释:把「字符串清理」和「数据验证」打包成独立的工具,再组合成流水线,优雅地处理脏数据。
💪 进阶 20 分钟:常见坑 + 性能小贴士
坑 1:闭包里的可变变量
# ❌ 错误示例
def create_counter():
count = 0
counters = []
def increment():
count += 1 # 这里会报错!因为count在increment里被当成新变量了
return count
counters.append(increment)
return counters
# ✅ 正确示例
def create_counter_fixed():
count = [0] # 用列表包装可变对象
counters = []
def increment():
count[0] += 1
return count[0]
counters.append(increment)
return counters
原理:Python 的 nonlocal 可以解决可变变量问题,但更简单的是用列表/字典包装。
坑 2:组合式函数被误当普通函数用
# ❌ 错误示例
result = create_counter() # 创建了实例但没用
counter = create_counter()
counter["increment"]() # 才是正确用法
# ✅ 正确示例:明确区分「工厂函数」和「使用」
def create_tracker():
"""工厂函数:生产追踪器实例"""
return {"track": lambda: print("追踪")}
坑 3:忘记 nonlocal 导致变量遮蔽
# ❌ 错误示例
count = 0
def create_counter():
count = 0 # 这是局部变量,不会修改外面的count
def increment():
count += 1 # UnboundLocalError
return count
# ✅ 正确示例
count = 0
def create_counter():
count = 0
def increment():
nonlocal count # 明确声明要修改外层变量
count += 1
return count
性能小贴士:缓存常用计算结果
def create_fibonacci():
"""斐波那契数列计算器(带缓存)"""
cache = {0: 0, 1: 1} # 缓存已计算过的值
def fib(n):
if n in cache:
return cache[n]
cache[n] = fib(n-1) + fib(n-2)
return cache[n]
def clear_cache():
cache.clear()
cache[0], cache[1] = 0, 1
return {"fib": fib, "clear_cache": clear_cache}
没有缓存的递归是 O(2^n),有缓存后是 O(n)。
调试技巧:用 vars() 查看闭包状态
def create_tracker():
clicks = 0
history = []
def track(event):
nonlocal clicks
clicks += 1
history.append(event)
return {"clicks": clicks, "history": history}
# 查看闭包里的变量
track.debug = lambda: {"clicks": clicks, "history": history.copy()}
return {"track": track}
t = create_tracker()
t["track"]("按钮点击")
t["track"]("页面滚动")
print(t["track"].debug()) # {'clicks': 2, 'history': ['按钮点击', '页面滚动']}
✏️ 练习题 + 作业题
练习题(5道,10分钟内完成)
练习 1(2分钟):创建你自己的计数器
- 输入:调用 increment() 3次,然后调用 value()
- 预期输出:3
- 提示:直接复制项目1的代码,改一下初始值试试
练习 2(2分钟):给待办清单加一个「优先级」
- 输入:add任务时传入优先级参数,filter按优先级过滤
- 预期输出:能按优先级筛选任务
- 提示:在todos字典里加一个 priority 字段
练习 3(2分钟):修复下面的错误
def create_printer():
msg = "Hello"
def print_msg():
msg = msg + " World" # 这行报错,怎么改?
print(msg)
- 预期输出:
Hello World - 提示:需要加一个关键字
练习 4(2分钟):组合两个已有的组合式函数
- 输入:用 create_click_tracker 和 create_validator 创建一个「带验证的点击追踪器」
- 预期输出:每次 track_click 都会先验证参数
- 提示:类似项目2的 create_form_handler 的写法
练习 5(2分钟):分析报错原因
def outer():
x = 10
def inner():
print(x) # 这里能访问到x吗?
x = 20 # 如果加了这行会怎样?
inner()
- 预期输出:分析并解释运行结果
- 提示:Python的变量作用域规则「LEGB」
作业题(30分钟-2小时)
作业:做一个「个人习惯追踪器」
- 需求描述:追踪你每天的习惯养成情况,比如喝水、运动、学习等
- 功能点:
1. 添加习惯(名称、目标次数)
2. 记录完成(每次完成+1)
3. 查看进度(今天完成了几/目标几次)
4. 统计报表(一周内的完成率) - 加分项:
1. 数据持久化(保存到JSON文件,重启不丢失)
2. 可视化输出(用 emoji 显示完成度 🎯📈) - 验收标准:
- 能跑起来
- 添加3个习惯,记录完成情况
- 打印出统计报表
- 代码有注释
- 提交方式:评论区贴代码或 GitHub 链接
📚 总结 + 资源
本文学到的3个核心点:
- 组合式函数 = 打包「状态+行为」,对外只暴露接口,内部状态私有
- 用
nonlocal修改外层变量,或者用列表/字典包装可变对象 - 组合式函数可以嵌套,像搭乐高一样组合出复杂功能
延伸学习资源:
- Python 官方文档:装饰器和闭包 —— 理解底层原理
- 《流畅的Python》第九章 —— 组合式设计模式的经典参考
- Real Python: Primer on Python Decorators —— 从装饰器角度理解闭包
互动钩子:
你在项目中有没有遇到过「这段代码到处复制粘贴」的痛苦?用了组合式函数之后是怎么解决的?评论区聊聊你的经历,老粉优先回复! 👇
📌 下章剧透:学会了封装可复用逻辑,下一章我们要用它来解决一个实际问题——怎么让用户在你的网页上「跳转」到不同页面?这就是 Vue Router 要教我们的东西……

评论(0)