第4章 4.4 Python 组合式编程高级用法

🎯 开场 3 分钟：为什么要学这个？

上一章我们学会了如何在插件市场里「淘宝」，找到了好几个省力的轮子。但你有没有这种感觉——用别人的插件虽然爽，可一旦要改点东西，就傻眼了？或者说，写出来的代码和自己写的业务搅在一起，改一行怕影响另一行？

举个真实的坑：你写了一个统计功能，开始还挺好，后来数据多了跑得越来越慢，你想把计算部分单独抽出来优化，结果发现计算逻辑和界面更新代码缠在一起，根本没法单独拎出来。

学完这一章，你能解决这些问题：

• 把一个复杂功能拆成独立的小块，想改哪块改哪块
• 代码复用更优雅，不用 Copy-Paste 三遍
• 写出跑得更快、占内存更少的数据处理代码
• 用 Python 也能写出「组合式」风格的代码

🧱 基础 25 分钟：核心概念

什么是「组合式」编程？

先说个生活例子。你去奶茶店点单，店员不是一次性做好所有东西，而是分步操作：先加茶底 → 加配料 → 加奶盖 → 封口。每一步都可以单独调整，换个配料不用重新做整杯。

组合式编\n\n\n\n\n\n程就是这个思路——把一个大功能拆成「可独立组合的小零件」。每个零件专注做一件事，用的时候像搭积木一样拼起来。

Python 的函数就是最好的零件。我们之前学的函数只能做固定操作，高级用法能让函数本身也「可定制」。

1. 高阶函数：函数界的「来料加工」

是什么：高阶函数就是「能接收函数作为输入，或者把函数作为输出」的函数。

类比：就像一个「来料加工店」，你把原材料（数据）送进去，店里可以根据你的要求（传入的函数）加工出不同产品。

# 这是一个高阶函数：把处理规则和数据分开
def process_items(items, process_func):
"""对列表中每个元素执行 process_func 处理"""
results = []
for item in items:
    results.append(process_func(item))
return results

# 定义两种不同的「加工规则」
def add_tax(price):
return price * 1.08

def with_discount(price):
return price * 0.9

# 同样的「来料加工店」，不同的加工规则
prices = [100, 200, 300]
print(process_items(prices, add_tax))      # [108.0, 216.0, 324.0]
print(process_items(prices, with_discount)) # [90.0, 180.0, 270.0]

这行 process_items(prices, add_tax) 说白了就是：「把价格列表送进去，按加税规则处理」。

2. 闭包：带「记忆」的函数

是什么：闭包是「记住出生时环境」的函数。就像你小时候养的狗，不管长大后去了哪里，它永远记得小时候你给它起的名字。

解决什么问题：有些参数我们不想每次调用都传，但函数内部需要用到。闭包帮你把这个参数「记住」。

def make_multiplier(factor):
"""创建一个「记住 factor」的乘法器"""
def multiplier(number):
    return number * factor  # factor 是「被记住」的环境变量
return multiplier

# 创建两个不同的乘法器
double = make_multiplier(2)   # 记住 factor=2
triple = make_multiplier(3)   # 记住 factor=3

# 同样调用 multiplier，结果不一样——因为记住的 factor 不同
print(double(5))   # 10  (5 * 2)
print(triple(5))   # 15  (5 * 3)

double 和 triple 是两个独立的函数，各自有各自的「记忆」。

3. 装饰器：函数的「包装盒」

是什么：装饰器就像快递包装。你给函数包一层外衣，不改变内容，但能加上新功能（防撞袋、加急标识等）。

类比：你去超市买水果，本来裸着的水果（函数），套上保鲜膜（装饰器），还是水果，但能保鲜、能标价。

import time

def timer(func):
"""装饰器：记录函数运行时间"""
def wrapper(*args, **kwargs):
    start = time.time()
    result = func(*args, **kwargs)
    end = time.time()
    print(f"{func.__name__} 跑了 {end - start:.4f} 秒")
    return result
return wrapper

@timer
def slow_function():
"""模拟一个慢函数"""
time.sleep(1)
return "完成"

@timer
def fast_function():
return "秒完"

# 使用装饰器：给函数包一层「计时」功能
slow_function()   # 输出：slow_function 跑了 1.0000 秒
fast_function()   # 输出：fast_function 跑了 0.0000 秒

注意！@timer 写在哪，Python 就自动把下面的函数包上这层「计时外衣」。

4. 生成器：数据的「自来水厂」

是什么：生成器不是一下子造好所有数据，而是「用多少产多少」。就像自来水厂，你开水龙头才来水，不用的时候不浪费。

解决什么问题：处理海量数据时，一次性加载到内存会爆掉。生成器让你「边用边取」，省内存。

def number_generator(n):
"""生成 0 到 n-1 的数字，但一次只生一个"""
for i in range(n):
    yield i  # yield 就是「产出一个，歇一会儿」

# 使用生成器
gen = number_generator(5)
print(next(gen))  # 0
print(next(gen))  # 1
print(next(gen))  # 2
print(list(gen))  # [3, 4]（剩下的全部取出）

# 更常用的方式：直接在循环里用
for num in number_generator(1000000):
print(num)  # 内存里永远只存一个数字，不会爆

5. 上下文管理器：资源的「保管箱」

是什么：上下文管理器帮你管理「打开→使用→关闭」的流程。就像停车场的闸机：入场抬杆（__enter__）、出场抬杆（__exit__），中间不管你停多久、停几辆，流程都是自动化的。

class FileManager:
"""自己写一个上下文管理器"""
def __init__(self, filename, mode):
    self.filename = filename
    self.mode = mode
    self.file = None

def __enter__(self):
    """进入时执行：打开文件"""
    self.file = open(self.filename, self.mode)
    return self.file

def __exit__(self, exc_type, exc_val, exc_tb):
    """退出时执行：关闭文件"""
    if self.file:
        self.file.close()
    return False  # 返回 False 表示不屏蔽异常

# 使用 with 语句，Python 自动帮你处理「打开/关闭」
with FileManager("test.txt", "w") as f:
f.write("Hello, World!")

# 出了 with 块，文件已经自动关闭，不用手动 f.close()

Python 还提供 contextlib 简化写法：

from contextlib import contextmanager

@contextmanager
def timer_context(label):
"""用装饰器语法创建上下文管理器"""
start = time.time()
yield label
end = time.time()
print(f"{label} 用了 {end - start:.4f} 秒")

with timer_context("下载文件"):
time.sleep(1)  # 模拟下载

6. 类型标注：代码的「产品说明书」

是什么：给变量、函数参数加个「类型标签」，让代码自己能「说话」，IDE 也能给你提建议。

from typing import List, Dict, Optional

def process_user_data(
users: List[Dict[str, str]],  # 用户列表，每个用户是「名字→年龄」的字典
filter_age: Optional[int] = None  # 可选的年龄过滤
) -> List[str]:  # 返回名字列表
"""处理用户数据，可按年龄过滤"""
results = []
for user in users:
    if filter_age is None or int(user.get("age", 0)) >= filter_age:
        results.append(user["name"])
return results

# 调用时 IDE 会提示你参数类型
users = [
{"name": "小明", "age": "25"},
{"name": "小红", "age": "17"},
{"name": "老王", "age": "35"}
]

print(process_user_data(users, 18))  # ['小明', '老王']

🔥 实战 35 分钟：3 个递进的小项目

项目 1（5 分钟）：待办清单 with 组合式函数

目标：理解高阶函数和闭包的实际用处。

"""待办清单 - 组合式版本"""

# 模拟数据存储
tasks = [
{"id": 1, "title": "写周报", "done": False, "priority": 2},
{"id": 2, "title": "开会", "done": True, "priority": 1},
{"id": 3, "title": "写代码", "done": False, "priority": 3},
]

def filter_tasks(tasks, condition_func):
"""通用的任务过滤函数（高阶函数）"""
return [task for task in tasks if condition_func(task)]

def mark_done(task):
"""标记完成"""
return {**task, "done": True}

# 使用闭包创建不同的「过滤规则」
is_pending = lambda t: not t["done"]
is_high_priority = lambda t: t["priority"] >= 3

# 组合使用
pending_tasks = filter_tasks(tasks, is_pending)
high_priority_pending = filter_tasks(pending_tasks, is_high_priority)

print("待完成的高优先级任务：")
for task in high_priority_pending:
print(f"  [{task['id']}] {task['title']}")

预期输出：

待完成的高优先级任务：
[3] 写代码

一句话解释：把「筛选逻辑」抽象成函数，代码更灵活，想换规则直接换函数就行。

项目 2（15 分钟）：CSV 数据清洗流水线

目标：从 CSV 读取数据，用生成器处理大数据，用装饰器加功能。

准备一个 sales.csv 文件：

date,product,sales,region
2024-01-01,手机,1200,华北
2024-01-01,电脑,3500,华东
2024-01-02,手机,980,华南
2024-01-02,电脑,4200,华北
2024-01-03,平板,2100,华东

代码：

"""CSV 数据清洗 - 生成器 + 装饰器版本"""

import csv
from typing import Generator, Dict

# 装饰器：给函数加「计时」功能
def log_calls(func):
"""记录函数被调用的次数"""
func.call_count = 0
def wrapper(*args, **kwargs):
    func.call_count += 1
    print(f"[{func.__name__} 调用 #{func.call_count}]")
    return func(*args, **kwargs)
return wrapper

@log_calls
def read_csv_generator(filename: str) -> Generator[Dict, None, None]:
"""生成器方式读取 CSV，一行一行吐出来，不占内存"""
with open(filename, "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        yield row

@log_calls
def clean_sales_data(row: Dict) -> Dict:
"""清洗一条销售记录"""
return {
    "date": row["date"],
    "product": row["product"].strip(),
    "sales": float(row["sales"]),

    "region": row["region"].strip()
}

@log_calls
def filter_by_region(row: Dict, region: str) -> bool:
"""过滤地区"""
return row["region"] == region

@log_calls
def calculate_total_sales(rows: Generator[Dict, None, None]) -> float:
"""计算总销售额"""
total = 0
for row in rows:
    total += row["sales"]
return total

# 主程序
print("=== 读取所有数据（生成器模式）===")
all_rows = read_csv_generator("sales.csv")

print("\n=== 清洗后的数据 ===")
cleaned = (clean_sales_data(row) for row in all_rows)  # 生成器管道

for item in cleaned:
print(f"  {item['date']} | {item['product']:4s} | ¥{item['sales']:>6.0f} | {item['region']}")

# 重新读取计算华东总额
print("\n=== 华东地区销售额 ===")
east_sales = (r for r in read_csv_generator("sales.csv"))
east_clean = (clean_sales_data(r) for r in east_sales)
east_filter = (r for r in east_clean if filter_by_region(r, "华东"))
print(f"华东总销售额：¥{calculate_total_sales(east_filter):.2f}")

预期输出：

=== 读取所有数据（生成器模式）===
[read_csv_generator 调用 #1]

=== 清洗后的数据 ===
[clean_sales_data 调用 #1]
2024-01-01 | 手机 |  ¥1200 | 华北
[clean_sales_data 调用 #2]
2024-01-01 | 电脑 |  ¥3500 | 华东
[clean_sales_data 调用 #3]
2024-01-02 | 手机 |   ¥980 | 华南
[clean_sales_data 调用 #4]
2024-01-02 | 电脑 |  ¥4200 | 华北
[clean_sales_data 调用 #5]
2024-01-03 | 平板 |  ¥2100 | 华东

=== 华东地区销售额 ===
[read_csv_generator 调用 #2]
[clean_sales_data 调用 #6]
[filter_by_region 调用 #1]
[clean_sales_data 调用 #7]
[filter_by_region 调用 #2]
[clean_sales_data 调用 #8]
华东总销售额：¥5600.00

一句话解释：用生成器做数据流水线，每个环节独立可替换，用多少取多少，内存不爆炸。

项目 3（15 分钟）：个人数据仪表盘

目标：综合使用装饰器、上下文管理器、类型标注，做一个可扩展的数据统计工具。

"""个人数据仪表盘 - 综合实战"""

from typing import List, Dict, Optional
from contextlib import contextmanager
import json

# ==================== 工具函数 ====================

@contextmanager
def timer(label: str = "操作"):
"""计时上下文管理器"""
import time
start = time.time()
yield
print(f"[{label}] 耗时 {time.time() - start:.3f}s")

def log_decorator(func):
"""日志装饰器"""
def wrapper(*args, **kwargs):
    print(f"\n>>> 开始执行：{func.__name__}")
    result = func(*args, **kwargs)
    print(f">>> 完成：{func.__name__}")
    return result
return wrapper

# ==================== 核心功能 ====================

@log_decorator
def load_json_context(filename: str) -> List[Dict]:
"""用上下文管理器安全读取 JSON"""
with open(filename, "r", encoding="utf-8") as f:
    return json.load(f)

@log_decorator
def analyze_expenses(expenses: List[Dict], min_amount: float = 0) -> Dict:
"""分析消费数据，返回统计结果"""
filtered = [e for e in expenses if e.get("amount", 0) >= min_amount]

total = sum(e["amount"] for e in filtered)
categories = {}

for expense in filtered:
    cat = expense.get("category", "其他")
    categories[cat] = categories.get(cat, 0) + expense["amount"]

return {
    "total": total,
    "count": len(filtered),
    "average": total / len(filtered) if filtered else 0,
    "by_category": categories
}

@log_decorator
def generate_report(stats: Dict, period: str = "本月") -> str:
"""生成文字报表"""
lines = [
    f"\n{'='*30}",
    f"📊 {period} 消费报告",
    f"{'='*30}",
    f"总消费：¥{stats['total']:.2f}",
    f"消费笔数：{stats['count']}",
    f"平均每笔：¥{stats['average']:.2f}",
    f"\n分类明细："
]

for cat, amount in sorted(stats["by_category"].items(), key=lambda x: -x[1]):
    pct = amount / stats['total'] * 100
    lines.append(f"  {cat}: ¥{amount:.2f} ({pct:.1f}%)")

return "\n".join(lines)

# ==================== 主程序 ====================

# 模拟数据（实际使用时从文件读取）
expense_data = [
{"date": "2024-01-15", "category": "餐饮", "amount": 45.5, "note": "午饭"},
{"date": "2024-01-16", "category": "交通", "amount": 8.0, "note": "地铁"},
{"date": "2024-01-17", "category": "餐饮", "amount": 128.0, "note": "请客"},
{"date": "2024-01-18", "category": "购物", "amount": 299.0, "note": "日用品"},
{"date": "2024-01-19", "category": "餐饮", "amount": 35.0, "note": "咖啡"},
]

if __name__ == "__main__":
print("💰 个人消费分析工具 v1.0\n")

with timer("完整分析流程"):
    # 分析所有消费
    stats = analyze_expenses(expense_data)

    # 生成报告
    report = generate_report(stats, "2024年1月中旬")
    print(report)

    # 分析高消费（>= 50元）
    print("\n" + "="*30)
    high_stats = analyze_expenses(expense_data, min_amount=50)
    high_report = generate_report(high_stats, "高消费分析（>=50元）")
    print(high_report)

预期输出：

💰 个人消费分析工具 v1.0

>>> 开始执行：analyze_expenses
>>> 完成：analyze_expenses

>>> 开始执行：generate_report
>>> 完成：generate_report

==============================
📊 2024年1月中旬 消费报告
==============================
总消费：¥515.50
消费笔数：5
平均每笔：¥103.10

分类明细：
餐饮: ¥208.50 (40.4%)
购物: ¥299.00 (58.0%)
交通: ¥8.00 (1.6%)

>>> 开始执行：analyze_expenses
>>> 完成：analyze_expenses

>>> 开始执行：generate_report
>>> 完成：generate_report

==============================
📊 高消费分析（>=50元）消费报告
==============================
总消费：¥471.50
消费笔数：3
平均每笔：¥157.17

分类明细：
餐饮: ¥173.00 (36.7%)
购物: ¥299.00 (63.3%)

[完整分析流程] 耗时 0.000s

一句话解释：把「计时」「日志」「数据读取」都做成独立组件，主程序只关心业务逻辑，改起来方便。

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：装饰器顺序搞反

# ❌ 错误示例：顺序搞反，效果不对
@timer
@log_decorator
def process():
pass

# ✅ 正确示例：先 log 再 timer
@log_decorator  # 外层：先记录日志
@timer          # 内层：再计时
def process():
pass

解释：装饰器从下往上执行，最近的那个最外层。

坑 2：生成器只能用一次

# ❌ 错误示例：生成器耗尽后再次使用
gen = (x for x in range(3))
print(list(gen))  # [0, 1, 2]
print(list(gen))  # [] —— 已经没了！

# ✅ 正确示例：需要多次使用就转成列表
data = [x for x in range(3)]  # 列表可以反复用
print(data)
print(data)

坑 3：装饰器忘记保留原函数信息

# ❌ 错误示例：wrapper 函数没有属性，原函数信息丢失
def bad_decorator(func):
def wrapper(*args):
    return func(*args)
return wrapper

@bad_decorator
def hello():
"""打招呼"""
pass

print(hello.__name__)  # 'wrapper' —— 不是 'hello' 了！
print(hello.__doc__)   # None —— 文档也没了

# ✅ 正确示例：用 wraps 保留原函数信息
from functools import wraps

def good_decorator(func):
@wraps(func)  # 关键这行！
def wrapper(*args):
    return func(*args)
return wrapper

@good_decorator
def hello():
"""打招呼"""
pass

print(hello.__name__)  # 'hello'
print(hello.__doc__)   # '打招呼'

坑 4：上下文管理器异常处理不当

# ❌ 错误示例：在 __exit__ 中 return True 屏蔽了异常
class BadManager:
def __exit__(self, exc_type, exc_val, exc_tb):
    self.close()
    return True  # 错误！异常被吞掉了

# ✅ 正确示例：只在需要屏蔽异常时返回 True
class GoodManager:
def __exit__(self, exc_type, exc_val, exc_tb):
    self.close()
    return False  # 异常正常传播

坑 5：类型标注用 list 而不是 List

# ❌ 错误示例：运行时类型提示不生效
def process(items: list):  # list 是类型，不是类型标注对象
pass

# ✅ 正确示例：使用 typing 模块
from typing import List, Dict, Tuple
def process(items: List[int], data: Dict[str, str]) -> Tuple[int, str]:
pass

性能小贴士：生成器链式处理大数据

# 场景：处理 1000 万行数据
import time

# ❌ 低效：先全部加载再处理
def bad_way():
data = [i for i in range(1000000)]  # 一次性加载到内存
filtered = [x for x in data if x % 2 == 0]
return sum(filtered)

# ✅ 高效：用生成器链式处理
def good_way():
def generate():
    for i in range(1000000):
        yield i

filtered = (x for x in generate() if x % 2 == 0)  # 永远是生成器
return sum(filtered)  # sum 边读边加，不占额外内存

# 测试
start = time.time()
bad_way()
print(f"低效方式: {time.time() - start:.3f}s")

start = time.time()
good_way()
print(f"高效方式: {time.time() - start:.3f}s")

调试技巧：用 logging 代替 print

import logging

# 设置日志级别和格式
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s [%(levelname)s] %(message)s"
)

def debug_demo():
data = [1, 2, 3, 4, 5]

logging.debug(f"输入数据: {data}")

filtered = [x for x in data if x > 2]
logging.debug(f"过滤后: {filtered}")

result = sum(filtered)
logging.info(f"计算结果: {result}")

debug_demo()

输出：

2024-01-20 10:30:00 [DEBUG] 输入数据: [1, 2, 3, 4, 5]
2024-01-20 10:30:00 [DEBUG] 过滤后: [3, 4, 5]
2024-01-20 10:30:00 [INFO] 计算结果: 12

✏️ 练习题

练习 1（1 分钟）：装饰器抄改

# 抄改下面代码，把 timer 装饰器加到 `greet` 函数上

def timer(func):
def wrapper(*args, **kwargs):
    import time
    start = time.time()
    result = func(*args, **kwargs)
    print(f"耗时 {time.time() - start:.2f}s")
    return result
return wrapper

@timer  # 把这行加到下面函数上
def greet(name):
import time
time.sleep(0.5)
return f"Hello, {name}!"

print(greet("小明"))

• 预期输出：Hello, 小明! 加上耗时提示

练习 2（2 分钟）：加 if 判断

# 在下面代码的 filter_tasks 中加一个判断：
# 只返回 priority >= 2 的任务

tasks = [
{"title": "写代码", "priority": 3},
{"title": "开会", "priority": 1},
{"title": "写周报", "priority": 2},
]

def filter_tasks(tasks, condition):
return [t for t in tasks if condition(t)]

# 运行这行，打印出 ["写代码", "写周报"]
print(filter_tasks(tasks, lambda t: # 在这里加条件))

练习 3（3 分钟）：用生成器处理新数据

# 把下面的列表转成生成器写法，输出前 3 个

numbers = [10, 20, 30, 40, 50]

# 改成生成器
def number_gen():
for n in numbers:
    yield n

gen = number_gen()
print(next(gen))
print(next(gen))
print(next(gen))

练习 4（4 分钟）：串起两个项目

# 用项目 2 的装饰器和项目 3 的上下文管理器组合

from contextlib import contextmanager

# 补全下面的代码：
# 1. 写一个 @log_calls 装饰器
# 2. 写一个 @contextmanager 的 timer
# 3. 在 process_data 函数上同时使用两者

def log_calls(func):
# 补全这里
pass

@contextmanager
def timer():
# 补全这里
yield

@log_calls
def process_data(data):
return [x * 2 for x in data]

with timer():
result = process_data([1, 2, 3])
print(result)

练习 5（挑战题，5 分钟）：报错分析

以下代码运行时报错，请分析原因并修复：

from functools import wraps

def memoize(func):
cache = {}
@wraps(func)
def wrapper(n):
    if n not in cache:
        cache[n] = func(n)
    return cache[n]
return wrapper

@memoize
def fibonacci(n):
if n < 2:
    return n
return fibonacci(n-1) + fibonacci(n-2)

# 报错信息：
# RecursionError: maximum recursion depth exceeded

print(fibonacci(100))  # 想计算 fibonacci(100)

• 提示：递归深度限制 + memoize 只对同一对象生效

作业：做一个「数据转换工具箱」

需求描述：
做一个命令行工具，可以对 CSV 文件进行「转换 → 过滤 → 统计」操作。

功能点：
1. 用生成器读取大文件（防止内存爆炸）
2. 支持按列过滤、按值过滤
3. 支持对数值列做统计（求和、平均、最大、最小）
4. 用装饰器实现「操作日志」和「计时」

加分项：
1. 支持导出结果到新 CSV
2. 支持管道式链式调用（transform().filter().aggregate()）

验收标准：
- 能处理 10000 行 CSV 不卡顿
- 输出符合预期的统计结果
- 代码有注释，关键函数有类型标注

📚 总结

本文学了 3 个核心点：
1. 高阶函数 + 闭包：把「规则」和数据分离，代码更灵活
2. 装饰器 + 上下文管理器：给函数加功能不改变核心逻辑
3. 生成器：处理大数据时不爆内存，边用边取

延伸资源：
- Python 官方文档「functools」模块：https://docs.python.org/zh-cn/3/library/functools.html
- 《流畅的 Python》- 第 7 章：装饰器和上下文管理器
- 视频：B 站「技术人办公桌」的 Python 进阶系列

你在处理数据时遇到过「内存不够」的情况吗？用的什么方法解决的？评论区聊聊，老粉优先回复！

下一章我们要做一个有点意思的综合实战——用组合式编程思想，做一个「数据转换流水线」，把学到的这些「积木」真正拼成一个实用工具。敬请期待！