一、🎭 装饰器：给函数穿上"魔法外衣"

想象一下，如果函数是演员，装饰器就是能瞬间更换的戏服，让演员不改变本身，却能展现出全新的能力和特性。

装饰器基本概念

装饰器是一个接收函数作为参数并返回一个新函数的高阶函数，它能在不修改原函数代码的情况下，增强原函数的功能。

# 最简单的装饰器
def simple_decorator(func):def wrapper():print("🌟 函数执行前")func()print("🌟 函数执行后")return wrapper# 使用装饰器
@simple_decorator
def say_hello():print("Hello, World!")# 调用被装饰的函数
say_hello()
# 输出:
# 🌟 函数执行前
# Hello, World!
# 🌟 函数执行后

这里的@simple_decorator语法糖等同于：

say_hello = simple_decorator(say_hello)

为装饰器添加参数传递功能

实际使用中，原函数通常有参数，我们需要确保装饰器能正确传递这些参数：

def smart_decorator(func):def wrapper(*args, **kwargs):print(f"🔍 调用函数: {func.__name__}")result = func(*args, **kwargs)print(f"✅ 函数返回: {result}")return resultreturn wrapper@smart_decorator
def add(a, b):return a + bprint(add(3, 5))  # 使用装饰器包装的函数
# 输出:
# 🔍 调用函数: add
# ✅ 函数返回: 8
# 8

带参数的装饰器

如果我们想让装饰器本身也能接收参数，需要再包装一层函数：

def repeat(times):def decorator(func):def wrapper(*args, **kwargs):result = Nonefor _ in range(times):result = func(*args, **kwargs)return resultreturn wrapperreturn decorator@repeat(times=3)
def greet(name):print(f"你好, {name}!")return "打招呼完成"greet("小明")
# 输出:
# 你好, 小明!
# 你好, 小明!
# 你好, 小明!

functools.wraps：保留原函数的元信息

默认情况下，装饰器会替换原函数，导致原函数的名称、文档字符串等元信息丢失。使用functools.wraps可以解决这个问题：

import functoolsdef my_decorator(func):@functools.wraps(func)  # 保留原函数的元信息def wrapper(*args, **kwargs):"""这是包装函数的文档"""print("Before function call")result = func(*args, **kwargs)print("After function call")return resultreturn wrapper@my_decorator
def example():"""这是原函数的文档"""print("Inside the function")print(example.__name__)  # 输出: example (而不是 wrapper)
print(example.__doc__)   # 输出: 这是原函数的文档

实用装饰器示例

1. 计时器装饰器

import time
import functoolsdef timer(func):@functools.wraps(func)def wrapper(*args, **kwargs):start_time = time.time()result = func(*args, **kwargs)end_time = time.time()print(f"⏱️ 函数 {func.__name__} 执行耗时: {end_time - start_time:.4f} 秒")return resultreturn wrapper@timer
def slow_function():time.sleep(1)return "函数执行完成"slow_function()
# 输出: ⏱️ 函数 slow_function 执行耗时: 1.0012 秒

2. 缓存装饰器(Memoization)

def memoize(func):cache = {}@functools.wraps(func)def wrapper(*args):if args in cache:print(f"💾 使用缓存结果: {args}")return cache[args]result = func(*args)cache[args] = resultprint(f"📝 缓存新结果: {args} -> {result}")return resultreturn wrapper@memoize
def fibonacci(n):if n <= 1:return nreturn fibonacci(n-1) + fibonacci(n-2)print(fibonacci(10))
# 首次计算会缓存所有中间结果
print(fibonacci(10))  # 第二次直接使用缓存

3. 权限检查装饰器

def require_auth(role):def decorator(func):@functools.wraps(func)def wrapper(user, *args, **kwargs):if user.get('role') != role:raise PermissionError(f"需要 {role} 权限!")return func(user, *args, **kwargs)return wrapperreturn decorator@require_auth(role="admin")
def delete_user(current_user, user_id):print(f"用户 {user_id} 已被删除")# 测试权限
admin_user = {'name': 'Admin', 'role': 'admin'}
normal_user = {'name': 'User', 'role': 'user'}delete_user(admin_user, 123)  # 正常执行
try:delete_user(normal_user, 123)  # 抛出权限错误
except PermissionError as e:print(f"错误: {e}")

4. 类方法装饰器

装饰器也可以用于类方法：

def log_method_calls(func):@functools.wraps(func)def wrapper(self, *args, **kwargs):print(f"📞 调用方法 {self.__class__.__name__}.{func.__name__}")return func(self, *args, **kwargs)return wrapperclass Calculator:def __init__(self, name):self.name = name@log_method_callsdef add(self, a, b):return a + b@log_method_callsdef multiply(self, a, b):return a * bcalc = Calculator("科学计算器")
calc.add(5, 3)      # 输出: 📞 调用方法 Calculator.add
calc.multiply(5, 3) # 输出: 📞 调用方法 Calculator.multiply

5. 类装饰器

装饰器不仅可以装饰函数，还可以装饰整个类：

def add_greeting(cls):# 给类添加一个新方法def say_hello(self):return f"{self.name} 说：你好!"cls.say_hello = say_helloreturn cls@add_greeting
class Person:def __init__(self, name):self.name = namep = Person("张三")
print(p.say_hello())  # 输出: 张三 说：你好!

二、🔄 迭代器：数据流的"传送带"

迭代器基本概念

迭代器是一种特殊的对象，它实现了迭代协议，允许我们逐个访问集合中的元素，而不需要知道集合的底层结构。

在Python中，迭代器需要实现两个方法：

• __iter__(): 返回迭代器对象本身
• __next__(): 返回下一个元素，如果没有更多元素则抛出StopIteration异常

# 简单迭代器示例
class CountDown:def __init__(self, start):self.current = startdef __iter__(self):# 返回迭代器对象自身return selfdef __next__(self):# 如果计数结束，抛出StopIterationif self.current <= 0:raise StopIteration# 否则返回当前值并递减value = self.currentself.current -= 1return value# 使用for循环遍历迭代器
for num in CountDown(5):print(num)
# 输出: 5 4 3 2 1# 手动使用迭代器
iterator = iter(CountDown(3))  # 调用__iter__()
print(next(iterator))  # 3     # 调用__next__()
print(next(iterator))  # 2
print(next(iterator))  # 1
# print(next(iterator))  # 抛出StopIteration异常

内置迭代器工具

Python的itertools模块提供了许多强大的迭代器工具：

import itertools# 无限迭代器
counter = itertools.count(1)  # 从1开始计数
print([next(counter) for _ in range(5)])  # [1, 2, 3, 4, 5]# 循环迭代器
cycle = itertools.cycle(["红", "黄", "绿"])
print([next(cycle) for _ in range(5)])  # ['红', '黄', '绿', '红', '黄']# 重复迭代器
repeat = itertools.repeat("A", 3)
print(list(repeat))  # ['A', 'A', 'A']# 链接多个迭代器
chain = itertools.chain([1, 2], [3, 4], [5, 6])
print(list(chain))  # [1, 2, 3, 4, 5, 6]# 分组迭代器
data = ["苹果", "梨", "菠萝", "葡萄", "芒果"]
for k, group in itertools.groupby(sorted(data), key=len):print(f"{k}个字符: {list(group)}")
# 输出:
# 1个字符: ['梨']
# 2个字符: ['苹果', '菠萝', '芒果', '葡萄']# 排列组合
print(list(itertools.combinations("ABC", 2)))  # [('A', 'B'), ('A', 'C'), ('B', 'C')]
print(list(itertools.permutations("ABC", 2)))  # [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]

🌱 生成器：懒惰的迭代器

生成器是一种特殊的迭代器，它使用函数的方式创建，通过yield关键字返回值，并保存函数的执行状态，等待下次调用。

生成器函数

使用yield关键字的函数会变成生成器函数，调用时返回一个生成器对象：

def countdown(n):print("开始倒计时!")while n > 0:yield n  # 返回当前值并暂停函数执行n -= 1# 创建生成器对象
generator = countdown(3)
print(type(generator))  # <class 'generator'># 迭代生成器
print(next(generator))  # 开始倒计时! 3
print(next(generator))  # 2
print(next(generator))  # 1
# print(next(generator))  # StopIteration异常# 使用for循环更方便
for num in countdown(3):print(num)
# 输出:
# 开始倒计时!
# 3
# 2
# 1

生成器表达式

类似于列表推导式，但使用圆括号而不是方括号：

# 列表推导式：立即计算所有结果并存储在内存中
squares_list = [x**2 for x in range(5)]
print(squares_list)  # [0, 1, 4, 9, 16]# 生成器表达式：按需计算
squares_gen = (x**2 for x in range(5))
print(squares_gen)  # <generator object <genexpr> at 0x...># 使用生成器
for square in squares_gen:print(square)  # 0 1 4 9 16

生成器的内存效率

生成器的主要优势是内存效率，特别是处理大数据集时：

import sys# 比较列表和生成器的内存使用
def get_size(obj):return sys.getsizeof(obj)# 创建一个大范围
big_range = 10**6# 使用列表
big_list = [x for x in range(big_range)]
print(f"列表大小: {get_size(big_list)} 字节")# 使用生成器
big_gen = (x for x in range(big_range))
print(f"生成器大小: {get_size(big_gen)} 字节")# 虽然两者都可以提供相同的数字序列，但内存占用差异巨大

生成器的进阶特性

1. 双向通信：send() 方法

生成器不仅可以产生值，还可以接收外部发送的值：

def echo_generator():while True:received = yield "等待输入..."  # yield一个值，然后等待send()的输入if received == 'exit':breakyield f"你说: {received}"  # 返回处理后的输入# 创建生成器
echo = echo_generator()
print(next(echo))  # 启动生成器: "等待输入..."print(echo.send("你好"))  # 发送值并获取下一个yield的值: "你说: 你好"
print(next(echo))  # "等待输入..."
print(echo.send("Python"))  # "你说: Python"
print(next(echo))  # "等待输入..."
echo.send("exit")  # 结束生成器

2. 生成器的异常处理：throw() 和 close()

生成器可以接收和处理外部抛入的异常：

def number_generator():try:for i in range(5):yield iexcept ValueError:print("捕获到ValueError!")yield "错误处理完毕"finally:print("生成器清理资源...")gen = number_generator()
print(next(gen))  # 0
print(next(gen))  # 1
print(gen.throw(ValueError))  # 抛出异常，输出: 捕获到ValueError! 错误处理完毕
gen.close()  # 关闭生成器，触发finally: 生成器清理资源...

3. 委托生成器：yield from

yield from允许一个生成器委托给另一个生成器：

def subgenerator(n):for i in range(n):yield i * idef delegating_generator():# 等同于迭代subgenerator并yield每个值yield from subgenerator(5)yield "子生成器完成"for value in delegating_generator():print(value)
# 输出: 0 1 4 9 16 子生成器完成

实用生成器示例

1. 文件读取生成器

逐行读取大文件而不将整个文件载入内存：

def read_large_file(file_path):with open(file_path, 'r') as file:for line in file:yield line.strip()# 使用示例
# for line in read_large_file("very_large_file.txt"):
#     process(line)

2. 无限素数生成器

def is_prime(n):"""检查一个数是否为素数"""if n <= 1:return Falseif n <= 3:return Trueif n % 2 == 0 or n % 3 == 0:return Falsei = 5while i * i <= n:if n % i == 0 or n % (i + 2) == 0:return Falsei += 6return Truedef infinite_primes():"""无限生成素数"""num = 2while True:if is_prime(num):yield numnum += 1# 获取前10个素数
prime_gen = infinite_primes()
first_10_primes = [next(prime_gen) for _ in range(10)]
print(first_10_primes)  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

3. 数据流处理管道

使用生成器构建数据处理管道：

def read_data():data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]for item in data:yield itemdef filter_even(items):for item in items:if item % 2 == 0:yield itemdef multiply_by_2(items):for item in items:yield item * 2def pipeline():# 构建处理管道result = multiply_by_2(filter_even(read_data()))return list(result)print(pipeline())  # [4, 8, 12, 16, 20]

三、🔗 将装饰器、迭代器和生成器结合使用

这些高级特性可以很好地结合使用，创建更强大的功能：

装饰生成器函数

def debug_generator(func):@functools.wraps(func)def wrapper(*args, **kwargs):gen = func(*args, **kwargs)for value in gen:print(f"🔍 生成器 {func.__name__} 产生: {value}")yield valuereturn wrapper@debug_generator
def counting(n):for i in range(n):yield i# 使用被装饰的生成器
for num in counting(3):print(f"使用值: {num}")
# 输出:
# 🔍 生成器 counting 产生: 0
# 使用值: 0
# 🔍 生成器 counting 产生: 1
# 使用值: 1
# 🔍 生成器 counting 产生: 2
# 使用值: 2

可重用的迭代器类

创建一个更复杂的可重用迭代器类：

class DataProcessor:def __init__(self, data):self.data = dataself.index = 0def __iter__(self):return selfdef __next__(self):if self.index >= len(self.data):raise StopIterationresult = self.process(self.data[self.index])self.index += 1return resultdef process(self, item):# 可在子类中重写的处理方法return item# 通过继承扩展功能
class SquareProcessor(DataProcessor):def process(self, item):return item ** 2# 使用迭代器
numbers = [1, 2, 3, 4, 5]
processor = SquareProcessor(numbers)
print(list(processor))  # [1, 4, 9, 16, 25]

四、 📊 实际应用案例：数据分析管道

结合所有概念创建一个数据处理管道：

import time
import functools# 装饰器：用于性能监控
def monitor(func):@functools.wraps(func)def wrapper(*args, **kwargs):start = time.time()result = func(*args, **kwargs)end = time.time()print(f"⏱️ {func.__name__} 处理耗时: {end - start:.4f}秒")return resultreturn wrapper# 生成器：数据读取
@monitor
def read_data(filename):"""模拟从文件读取数据"""print(f"📂 从{filename}读取数据...")# 模拟数据data = [{"id": 1, "name": "Product A", "price": 100, "category": "Electronics"},{"id": 2, "name": "Product B", "price": 50, "category": "Clothing"},{"id": 3, "name": "Product C", "price": 150, "category": "Electronics"},{"id": 4, "name": "Product D", "price": 80, "category": "Home"},{"id": 5, "name": "Product E", "price": 200, "category": "Electronics"}]time.sleep(0.1)  # 模拟I/O延迟for item in data:yield item# 生成器：数据过滤
@monitor
def filter_data(items, category):"""筛选特定类别的商品"""print(f"🔍 筛选{category}类别...")for item in items:if item["category"] == category:yield item# 生成器：数据转换
@monitor
def transform_data(items):"""计算商品价格的销售价(打八折)"""print("🔄 计算销售价...")for item in items:transformed = item.copy()transformed["sale_price"] = item["price"] * 0.8yield transformed# 消费生成器：保存结果
@monitor
def save_results(items, output_filename):"""保存处理后的数据"""print(f"💾 保存结果到{output_filename}...")results = list(items)  # 消耗生成器# 模拟保存操作time.sleep(0.1)print(f"✅ 已保存{len(results)}条记录")return results# 构建完整的数据处理管道
def process_sales_data(input_filename, output_filename, category):"""完整的数据处理流程"""# 构建处理管道data = read_data(input_filename)filtered_data = filter_data(data, category)transformed_data = transform_data(filtered_data)results = save_results(transformed_data, output_filename)# 打印结果示例if results:print("\n📊 处理结果示例:")for item in results[:2]:  # 只显示前两条print(f"  {item['name']}: 原价¥{item['price']}, 销售价¥{item['sale_price']:.2f}")if len(results) > 2:print(f"  ...以及其他{len(results)-2}条记录")# 执行数据处理管道
process_sales_data("products.csv", "electronics_sales.csv", "Electronics")

总结

1. 装饰器

• 允许在不修改原函数的情况下添加新功能
• 可以用于函数或类
• 实用场景：日志记录、性能监控、访问控制、缓存等

2. 迭代器

• 实现了__iter__和__next__方法的对象
• 允许逐个访问集合元素，而不需要加载整个集合
• Python内置了丰富的迭代器工具(itertools)

3. 生成器

• 使用yield语句的特殊函数
• 执行时会保存状态，按需生成值
• 极大减少内存使用，适合处理大数据集
• 高级特性：send(), throw(), close(), yield from

这三者是Python高级编程中的关键概念，掌握它们可以编写出更优雅、高效和可维护的代码。尤其在处理大量数据、构建数据处理管道或需要分布式计算时，这些概念尤为重要。

练习建议

1. 构建装饰器库

   • 创建一组实用装饰器：计时、重试、缓存等
   • 尝试组合多个装饰器

2. 设计自定义迭代器

   • 实现一个模拟数据库游标的迭代器
   • 创建一个分页迭代器，按批次返回数据

3. 生成器应用

   • 实现一个大文件处理程序，使用生成器进行内存高效处理
   • 构建数据转换管道，将数据从一种格式转换为另一种格式

4. 综合项目

   • 开发一个简单的ETL(提取-转换-加载)工具
   • 设计一个支持链式操作的数据处理框架

🚀 下一步学习：探索Python的函数式编程特性