Python装饰器、生成器、上下文管理器：3个必会的"魔法"机制
【摘要】 Python装饰器、生成器、上下文管理器：3个必会的"魔法"机制关键词：Python进阶、装饰器、生成器、上下文管理器、with语句、yield、闭包一、引言：为什么是这三个？如果你已经写了一段时间的Python，大概率见过这样的代码：@timerdef process_data(): ...for chunk in read_large_file("huge.csv"): .....
Python装饰器、生成器、上下文管理器：3个必会的"魔法"机制
关键词：Python进阶、装饰器、生成器、上下文管理器、with语句、yield、闭包
一、引言：为什么是这三个？
如果你已经写了一段时间的Python，大概率见过这样的代码：
@timer
def process_data():
...
for chunk in read_large_file("huge.csv"):
...
with open("data.txt") as f:
content = f.read()这三段代码分别用到了装饰器、生成器和上下文管理器。
它们看起来像"语法糖"，但实际上，它们是Python设计哲学的核心体现——把复杂留给自己，把简洁留给用户。
这篇文章的目标：从"会用"到"理解"，再到"能自己写"。
二、装饰器（Decorator）：给函数"穿衣服"
2.1 什么是装饰器？
一句话：装饰器是一个接收函数、返回函数的函数。
它的作用是在不修改原函数代码的前提下，给函数增加新功能。
打个比方：你有一部手机（原函数），你想防水，但不想拆开手机改装。怎么办？买个防水壳套上去。这个防水壳就是装饰器。
2.2 从最朴素的需求开始
假设你想统计一个函数的执行时间：
import time
def slow_function():
time.sleep(1)
print("Done!")
# 朴素写法：在函数里加代码
def slow_function():
start = time.time()
time.sleep(1)
print("Done!")
end = time.time()
print(f"耗时: {end - start:.2f}秒")问题来了：如果你有20个函数都要统计时间，难道每个都改一遍？
2.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:.2f}秒")
return result
return wrapper
# 使用方式一：函数式调用
def slow_function():
time.sleep(1)
print("Done!")
slow_function = timer(slow_function)
slow_function()输出：
Done!
slow_function 耗时: 1.00秒2.4 语法糖：@符号
上面那种
slow_function = timer(slow_function) 的写法太丑了。Python提供了 @ 语法糖：@timer
def slow_function():
time.sleep(1)
print("Done!")
slow_function()@timer 就等于 slow_function = timer(slow_function)。就这么简单。
2.5 带参数的装饰器
如果我想让装饰器本身也能接收参数呢？比如控制日志级别：
def log(level="INFO"):
def decorator(func):
def wrapper(*args, **kwargs):
print(f"[{level}] 调用 {func.__name__}")
return func(*args, **kwargs)
return wrapper
return decorator
@log(level="DEBUG")
def greet(name):
print(f"Hello, {name}!")
greet("Alice")[DEBUG] 调用 greet
Hello, Alice!三层嵌套，从外到内：
log(level) → 返回 decorator
decorator(func) → 返回 wrapper
wrapper(...) → 执行增强逻辑 + 原函数2.6 保留原函数信息
你可能会发现，装饰后的函数"丢失"了原函数的名字和文档：
@timer
def slow_function():
"""这是一个很慢的函数"""
time.sleep(1)
print(slow_function.__name__) # 输出: wrapper ← 不对！
print(slow_function.__doc__) # 输出: None ← 不对！解决办法：用
functools.wraps：import functools
def timer(func):
@functools.wraps(func) # ← 加上这行
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} 耗时: {end - start:.2f}秒")
return result
return wrapper2.7 实用场景
|
场景
|
装饰器示例
|
|---|---|
|
性能统计
|
@timer |
|
权限校验
|
@login_required |
|
缓存结果
|
@functools.lru_cache |
|
重试机制
|
@retry(max_attempts=3) |
|
路由注册
|
@app.route("/home") (Flask) |
2.8 内置装饰器
Python自带了几个常用的：
# 缓存：同样的参数只计算一次
@functools.lru_cache(maxsize=128)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
# 把方法变成属性调用
class Circle:
def __init__(self, radius):
self.radius = radius
@property
def area(self):
return 3.14159 * self.radius ** 2
c = Circle(5)
print(c.area) # 像属性一样调用，不用加括号
# 类方法
class MyClass:
@classmethod
def from_string(cls, s):
...
@staticmethod
def helper():
...三、生成器（Generator）：懒惰是美德
3.1 什么是生成器？
一句话：生成器是一个"按需生产"的可迭代对象。
普通函数用
return 一次性返回所有结果；生成器用 yield 每次返回一个结果，然后"暂停"，等你需要下一个时再"恢复"。3.2 从列表的问题说起
# 生成1到100万的平方
def get_squares(n):
result = []
for i in range(n):
result.append(i ** 2)
return result
squares = get_squares(1_000_000) # 内存中存了100万个数字问题：如果只需要遍历一次，这100万个数字占用了大量内存（约8MB），而且你可能在第10个就break了，后面999990个白算了。
3.3 用生成器改写
def get_squares_gen(n):
for i in range(n):
yield i ** 2 # ← 关键：yield不是return
# 使用
gen = get_squares_gen(1_000_000)
print(gen) # <generator object ...>
for sq in gen:
print(sq)
if sq > 100:
break区别：
|
普通函数
|
生成器
|
|
|---|---|---|
|
返回方式
|
return |
yield |
|
返回次数
|
一次
|
多次
|
|
内存占用
|
存所有结果
|
只存当前值
|
|
执行方式
|
从头跑到尾
|
跑到yield暂停，下次从暂停处继续
|
3.4 生成器的工作原理
def counter():
print("开始")
yield 1
print("继续")
yield 2
print("结束")
yield 3
c = counter()
print(next(c)) # 输出: 开始 \n 1
print(next(c)) # 输出: 继续 \n 2
print(next(c)) # 输出: 结束 \n 3
print(next(c)) # StopIteration 异常关键理解：
- 调用
counter()时，函数体并没有执行 - 第一次
next(c)，执行到第一个yield，返回1，暂停 - 第二次
next(c)，从暂停处继续，执行到第二个yield，返回2，暂停 - 以此类推，直到没有更多
yield，抛出StopIteration
3.5 生成器表达式
和列表推导式很像，只是把
[] 换成 ()：# 列表推导式
squares_list = [x**2 for x in range(1000000)] # 立即生成所有
# 生成器表达式
squares_gen = (x**2 for x in range(1000000)) # 按需生成3.6 用生成器处理大文件
这是生成器最经典的场景：
def read_large_file(filepath, chunk_size=8192):
"""逐行读取大文件，不占用大量内存"""
with open(filepath, 'r') as f:
for line in f:
yield line.strip()
# 处理10GB的日志文件
for line in read_large_file("huge_log.txt"):
if "ERROR" in line:
print(line)3.7 管道式数据处理
生成器可以串联，形成处理管道：
def read_lines(filepath):
with open(filepath) as f:
for line in f:
yield line.strip()
def filter_error(lines):
for line in lines:
if "ERROR" in line:
yield line
def extract_time(lines):
for line in lines:
yield line.split()[0] # 假设时间在第一列
# 管道串联
lines = read_lines("app.log")
errors = filter_error(lines)
times = extract_time(errors)
for t in times:
print(t)每个生成器都是惰性的，数据像流水线一样逐条处理，内存占用极低。
3.8 send() 和 yield 的高级用法
生成器不仅能"产出"值，还能"接收"值：
def accumulator():
total = 0
while True:
value = yield total # 产出total，同时接收value
if value is None:
break
total += value
acc = accumulator()
next(acc) # 启动生成器，执行到第一个yield
print(acc.send(10)) # 发送10，total=10，产出10
print(acc.send(20)) # 发送20，total=30，产出30
print(acc.send(5)) # 发送5，total=35，产出35这是协程（coroutine）的基础，也是
async/await 的前身。四、上下文管理器（Context Manager）：善始善终
4.1 什么是上下文管理器？
一句话：上下文管理器保证你"进门时开灯，出门时关灯"。
最常见的用法就是
with 语句。4.2 从资源泄漏说起
# 打开文件但不关闭——资源泄漏！
f = open("data.txt", "r")
content = f.read()
# 如果这里抛异常了，f.close()永远不会执行
f.close()改进版：
f = open("data.txt", "r")
try:
content = f.read()
finally:
f.close() # 确保关闭写起来啰嗦。用
with：with open("data.txt", "r") as f:
content = f.read()
# 文件自动关闭，即使里面抛异常4.3 自己写一个上下文管理器
方法一：类实现（enter 和 exit）
class Timer:
def __enter__(self):
self.start = time.time()
return self # ← with语句中as后面的就是这个返回值
def __exit__(self, exc_type, exc_val, exc_tb):
self.end = time.time()
print(f"耗时: {self.end - self.start:.2f}秒")
# 返回False则异常继续传播，返回True则吞掉异常
return False
# 使用
with Timer():
time.sleep(1)方法二：用 @contextmanager 装饰器（更简洁）
from contextlib import contextmanager
@contextmanager
def timer():
start = time.time()
yield # ← yield之前的代码相当于__enter__
end = time.time() # ← yield之后的代码相当于__exit__
print(f"耗时: {end - start:.2f}秒")
# 使用
with timer():
time.sleep(1)对比：
|
方式
|
代码量
|
灵活性
|
适合场景
|
|---|---|---|---|
|
类实现
|
多
|
高（可处理异常细节）
|
复杂资源管理
|
|
@contextmanager
|
少
|
中
|
简单场景
|
4.4 实用场景
场景一：临时切换工作目录
@contextmanager
def cd(path):
old_dir = os.getcwd()
os.chdir(path)
try:
yield
finally:
os.chdir(old_dir)
with cd("/tmp"):
# 在/tmp目录下工作
print(os.getcwd())
# 自动回到原目录场景二：数据库连接
@contextmanager
def get_db_connection():
conn = psycopg2.connect("dbname=test user=postgres")
try:
yield conn
finally:
conn.close()
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")场景三：抑制异常
from contextlib import suppress
# 不想让FileNotFoundError打断程序
with suppress(FileNotFoundError):
os.remove("temp.txt")
# 文件不存在也不报错4.5 contextlib 工具箱
from contextlib import closing, redirect_stdout, ExitStack
# closing：给没有close()的对象加上上下文管理
from urllib.request import urlopen
with closing(urlopen("http://example.com")) as page:
content = page.read()
# redirect_stdout：临时重定向输出
with redirect_stdout(open("output.txt", "w")):
print("这段文字写到了文件里")
# ExitStack：动态管理多个上下文
with ExitStack() as stack:
files = [stack.enter_context(open(f"file{i}.txt")) for i in range(5)]
# 5个文件全部自动关闭五、三者结合：一个完整案例
把装饰器、生成器、上下文管理器放在一起，写一个带性能监控的大文件处理工具：
import time
import functools
from contextlib import contextmanager
# ========== 装饰器：性能监控 ==========
def monitor(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
with timer(): # ← 上下文管理器
return func(*args, **kwargs)
return wrapper
# ========== 上下文管理器：计时 ==========
@contextmanager
def timer():
start = time.time()
yield
end = time.time()
print(f"⏱️ [{timer.__module__}] 总耗时: {end - start:.2f}秒")
# ========== 生成器：逐行读取 ==========
def read_chunks(filepath, chunk_size=1024):
with open(filepath, 'r') as f: # ← 上下文管理器
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk # ← 生成器
# ========== 组合使用 ==========
@monitor
def count_words_in_large_file(filepath):
"""统计大文件中的单词数"""
total_words = 0
for chunk in read_chunks(filepath): # ← 消费生成器
total_words += len(chunk.split())
return total_words
# 使用
result = count_words_in_large_file("huge_text.txt")
print(f"单词总数: {result}")输出：
⏱️ 总耗时: 2.34秒
单词总数: 158392这个例子展示了什么？
@monitor装饰器给函数加了计时能力，不需要改函数内部read_chunks生成器让大文件处理不爆内存with open()上下文管理器确保文件正确关闭- 三者各司其职，组合在一起优雅而强大
六、对比总结
|
机制
|
核心关键字
|
本质
|
解决的问题
|
|---|---|---|---|
|
装饰器
|
@ |
函数嵌套
|
不修改原代码增加功能
|
|
生成器
|
yield |
暂停/恢复
|
大数据集的惰性处理
|
|
上下文管理器
|
with |
资源生命周期
|
确保资源正确释放
|
心智模型
┌─────────────────────────────────────────────┐
│ 装饰器 = 给函数"套壳" │
│ ┌─────────┐ │
│ │ 新功能 │ ← 计时、日志、权限... │
│ │┌────────│────────┐ │
│ ││ 原函数 │ │ │
│ │└────────│────────┘ │
│ └─────────┘ │
├─────────────────────────────────────────────┤
│ 生成器 = 工厂流水线 │
│ │
│ 原料 → [加工①] → 产出① → 暂停 │
│ [加工②] → 产出② → 暂停 │
│ [加工③] → 产出③ → 暂停 │
│ ... │
│ 不是一次性生产所有产品，而是按需生产 │
├─────────────────────────────────────────────┤
│ 上下文管理器 = 自动门 │
│ │
│ 进门(enter) → 你在里面做事 → 出门(exit) │
│ 不管你做没做完、有没有摔倒，门都会自动关 │
└─────────────────────────────────────────────┘七、学习路径建议
|
阶段
|
目标
|
练习
|
|---|---|---|
|
入门
|
能看懂别人的装饰器/生成器/with
|
阅读标准库源码（如
functools.lru_cache） |
|
进阶
|
能写简单的装饰器和生成器
|
写一个
@retry装饰器、写一个斐波那契生成器 |
|
熟练
|
能灵活组合三者
|
写一个带缓存的、可计时的、逐行处理文件的工具
|
|
精通
|
理解底层原理
|
研究Python的
__call__、gi_frame、__exit__的异常处理机制 |
八、结语
这三个"魔法"机制，本质上都是Python对关注点分离的实践：
- 装饰器把"做什么"和"怎么做之前/之后"分离
- 生成器把"数据生产"和"数据消费"分离
- 上下文管理器把"资源获取"和"资源释放"分离
理解了这个本质，你就不会再觉得它们是"魔法"，而是顺理成章的工具。
最好的学习方式就是动手写。 把这篇文章里的每个例子都敲一遍，改一改，跑一跑。当你能自然地用
@timer 给函数计时、用 yield 处理数据流、用 with 管理资源时，你就已经迈入了Python进阶的大门。
【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）