第9章 9.2 装饰器从原理到实战
🎯 开场:快递盒子的启示
上一章我们学会了列表推导式,用一条简洁的「点菜流水线」把繁琐的循环变成了优雅的一行代码。如果你已经能熟练用它来清洗数据,恭喜你——你已经从「会写代码」往「写好代码」迈了一大步。
但今天我要教你一个更神奇的工具,它能让你不动原代码就给函数加新功能。听起来很玄?听我讲个故事。
你有没有遇到过这种情况:写了一个函数,运行正常,但某天产品经理说「加上日志吧」「加上性能计时吧」「加上权限校验吧」——你一加,发现整个代码结构要改,或者好几个地方都要复制粘贴同样的代码。
痛点来了:想给函数加个通用功能,却要改函数本身,重复代码满天飞。
这就是今天要解决的——装饰器。学会了它,上面那些需求,一行代码搞定。
🧱 基础 25 分钟:核心概念
什么是装饰器?说白了就是「包装纸」
想象你有一个快递盒子(函数),你想在外面包一层气泡垫(额外功能),但又不想把盒子里的东西(函数本身)拆开重装。装饰器就是这个思路——在函数外面包一层包装,不动原函数就给它加功能。
第一步:理解闭包——装饰器的「内功心法」
装饰器的底层原理是闭包。闭包是啥?
生活类比:你把一本书(变量)放在包里(函数),然后把包借给别人。别人虽然拿不到书,但下次他打开包的时候,还能看到那本书在里头。
用代码说话:
def 外层函数():
收藏的古董 = "1980年的邮票" # 外部变量
def 内层函数():
print(f"我看到了:{收藏的古董}") # 引用外部变量
return 内层函数 # 把内层函数当成返回值
拿到的东西 = 外层函数()
拿到的东西() # 输出:我看到了:1980年的邮票
内层函数记住了外层函数的变量,这就是闭包。闭包让内部函数能「记住」定义时的环境。

第二步:写第一个装饰器
为什么要用:不动原函数代码,给它加上新功能。
怎么用:定义一个「包装函数」,把被装饰的函数包进去。
def my_decorator(func): # func是被装饰的函数
def wrapper(): # wrapper是包装层
print("我是包装纸,在函数执行前来报到!")
func() # 调用原函数
print("我是包装纸,在函数执行后也来报到!")
return wrapper
@my_decorator # 这就是装饰器的写法
def say_hello():
print("Hello!")
say_hello()
输出:
我是包装纸,在函数执行前来报到!
Hello!
我是包装纸,在函数执行后也来报到!
解释:@my_decorator这行代码干了三件事:1)调用my_decorator并把say_hello传进去;2)得到返回值(wrapper函数);3)用这个wrapper替代原来的say_hello。
第三步:@functools.wraps——保留原函数的「身份信息」
问题来了:装饰器包装之后,原函数的元信息(名字、文档字符串)都丢了。
def my_decorator(func):
def wrapper():
print("执行前")
func()
print("执行后")
return wrapper
@my_decorator
def say_hello():
"""这是一个问候函数"""
print("Hello!")
print(say_hello.__name__) # 输出:wrapper(不对!应该输出say_hello)
print(say_hello.__doc__) # 输出:None(也不对!)
解决方案:functools.wraps
import functools
def my_decorator(func):
@functools.wraps(func) # 保持原函数的身份
def wrapper():
print("执行前")
func()
print("执行后")
return wrapper
@my_decorator
def say_hello():
"""这是一个问候函数"""
print("Hello!")
print(say_hello.__name__) # 输出:say_hello
print(say_hello.__doc__) # 输出:这是一个问候函数
注意:@functools.wraps(func)要写在wrapper函数上面,顺序不能错。

第四步:带参数的装饰器——给包装纸加开口
如果被装饰的函数有参数怎么办?
import functools
def my_decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs): # 接收任意参数
print("开始执行...")
result = func(*args, **kwargs) # 把参数原封不动传给原函数
print("执行结束")
return result # 返回原函数的返回值
return wrapper
@my_decorator
def add(a, b):
"""求和函数"""
return a + b
print(add(3, 5))
输出:
开始执行...
执行结束
8
*args和**kwargs的意思是「不管你传什么参数,我都接着,然后原封不动传给原函数」。这就像快递公司说「不管你的盒子多大,我都给你包气泡垫」。
第五步:装饰器带参数——定制不同款式的包装纸
有时候需要让装饰器本身也能接受参数,比如「日志装饰器」要指定日志级别。
import functools
def log(level): # 先接受装饰器的参数
def decorator(func): # 再接受被装饰的函数
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"[{level}] 开始执行 {func.__name__}")
result = func(*args, **kwargs)
print(f"[{level}] 执行完毕")
return result
return wrapper
return decorator
@log("INFO") # 指定日志级别
def add(a, b):
return a + b
add(1, 2)
输出:
[INFO] 开始执行 add
[INFO] 执行完毕
这个@log("INFO")其实是两步:先调用log("INFO")得到decorator,再把add函数传进去。
第六步:类装饰器——用类做包装纸
装饰器不仅可以用函数实现,还可以用类。适合需要「保存状态」的场景。
import functools
class CountCalls:
def __init__(self, func):
self.count = 0
functools.update_wrapper(self, func) # 保留原函数信息
def __call__(self, *args, **kwargs): # 让实例能像函数一样被调用
self.count += 1
print(f"这个函数被调用了 {self.count} 次")
return self._func(*args, **kwargs)
def _func(self, *args, **kwargs):
# 这里需要用另一个方式保存原函数,见下方正确写法
pass
# 正确写法:
class CountCalls:
def __init__(self, func):
self._func = func
self.count = 0
functools.update_wrapper(self, func)
def __call__(self, *args, **kwargs):
self.count += 1
print(f"调用次数:{self.count}")
return self._func(*args, **kwargs)
@CountCalls
def say_hello():
print("Hello!")
say_hello() # 输出:调用次数:1,然后 Hello!
say_hello() # 输出:调用次数:2,然后 Hello!
类装饰器的核心:__call__方法让类的实例可以像函数一样被调用。
🔥 实战 35 分钟:3 个递进的小项目
项目 1(5 分钟):计时装饰器
场景:你想知道某个函数的执行时间。
import functools
import time
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:.4f} 秒")
return result
return wrapper
@timer
def slow_function():
time.sleep(0.5)
return "完成"
@timer
def quick_add(a, b):
return a + b
print(slow_function())
print(quick_add(1, 2))
预期输出:
slow_function 执行耗时:0.5003 秒
完成
quick_add 执行耗时:0.0000 秒
3
一句话解释:time.time()记录起止时间,相减得出耗时。
项目 2(15 分钟):日志记录器——记录函数调用情况
场景:你写了一个数据处理脚本,需要记录每次处理了哪些数据、出错了吗、用了多久。
import functools
import time
import json
def logger(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
log_entry = {
"函数": func.__name__,
"调用时间": time.strftime("%Y-%m-%d %H:%M:%S"),
"参数": str(args) if args else "无",
"关键字参数": str(kwargs) if kwargs else "无"
}
print(f"[LOG] 调用 {func.__name__}")
try:
result = func(*args, **kwargs)
log_entry["状态"] = "成功"
log_entry["返回值"] = str(result)
return result
except Exception as e:
log_entry["状态"] = "失败"
log_entry["错误"] = str(e)
raise
finally:
# 实际项目中可以写入文件
print(f"[LOG] {json.dumps(log_entry, ensure_ascii=False)}")
return wrapper
@logger
def process_order(order_id, amount, vip=False):
"""模拟订单处理"""
print(f" 正在处理订单 {order_id},金额 {amount} 元")
if amount > 1000:
raise ValueError("金额超标!")
return {"order_id": order_id, "processed": True}
# 测试正常情况
try:
result = process_order("ORD001", 500, vip=True)
print(f"结果:{result}")
except Exception as e:
print(f"出错了:{e}")
print("-" * 30)
# 测试异常情况
try:
result = process_order("ORD002", 2000)
except Exception as e:
print(f"预期中的错误:{e}")
预期输出:
[LOG] 调用 process_order
正在处理订单 ORD001,金额 500 元
[LOG] {"函数": "process_order", "调用时间": "...", "参数": "('ORD001', 500)", "关键字参数": "{'vip': True}", "状态": "成功", "返回值": "{'order_id': 'ORD001', 'processed': True}"}
结果:{'order_id': 'ORD001', 'processed': True}
------------------------------
[LOG] 调用 process_order
正在处理订单 ORD002,金额 2000 元
[LOG] {"函数": "process_order", "调用时间": "...", "参数": "('ORD002', 2000)", "关键字参数": "{}", "状态": "失败", "错误": "金额超标!"}
预期中的错误:金额超标!
一句话解释:try/except/finally确保成功失败都能记录日志。
项目 3(15 分钟):组合技能——做一个带缓存的爬虫装饰器
场景:写爬虫时,重复请求相同的URL很浪费,可以用缓存把结果存下来。
import functools
import time
def cache(func):
"""缓存装饰器,相同的参数直接返回缓存结果"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 用参数生成缓存key
cache_key = str(args) + str(kwargs)
if not hasattr(wrapper, '_cache'):
wrapper._cache = {}
if cache_key in wrapper._cache:
print(f"[缓存命中] {func.__name__}{args}")
return wrapper._cache[cache_key]
print(f"[缓存未命中] {func.__name__}{args}")
result = func(*args, **kwargs)
wrapper._cache[cache_key] = result
return result
return wrapper
def retry(max_attempts=3, delay=1):
"""重试装饰器,失败时自动重试"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt < max_attempts - 1:
print(f"[重试] {func.__name__} 失败,{delay}秒后重试...")
time.sleep(delay)
else:
print(f"[重试失败] {func.__name__} 最终失败:{e}")
raise
return wrapper
return decorator
@cache
@retry(max_attempts=3, delay=0.5)
def fetch_data(url):
"""模拟爬虫获取数据"""
# 模拟网络请求
time.sleep(0.5)
data = {
"https://api.example.com/users": ["Alice", "Bob", "Charlie"],
"https://api.example.com/products": ["笔记本", "手机", "平板"]
}
if url not in data:
raise ValueError(f"URL不存在:{url}")
return {"url": url, "data": data[url], "timestamp": time.time()}
# 第一次调用(缓存未命中)
print("=== 第1次调用 ===")
result1 = fetch_data("https://api.example.com/users")
print(f"结果:{result1}")
# 第二次调用(缓存命中)
print("\n=== 第2次调用 ===")
result2 = fetch_data("https://api.example.com/users")
print(f"结果:{result2}")
# 第三次调用不同URL
print("\n=== 第3次调用(新URL) ===")
result3 = fetch_data("https://api.example.com/products")
print(f"结果:{result3}")
预期输出:
=== 第1次调用 ===
[缓存未命中] fetch_data('https://api.example.com/users',)
结果:{'url': 'https://api.example.com/users', 'data': ['Alice', 'Bob', 'Charlie'], 'timestamp': ...}
=== 第2次调用 ===
[缓存命中] fetch_data('https://api.example.com/users',)
结果:{'url': 'https://api.example.com/users', 'data': ['Alice', 'Bob', 'Charlie'], 'timestamp': ...}
=== 第3次调用(新URL) ===
[缓存未命中] fetch_data('https://api.example.com/products',)
结果:{'url': 'https://api.example.com/products', 'data': ['笔记本', '手机', '平板'], 'timestamp': ...}
一句话解释:两个装饰器叠加,@cache在外层先检查缓存,@retry在内层处理失败重试。
💪 进阶 20 分钟:常见坑 + 调试技巧
坑 1:装饰器顺序搞反
# ❌ 错误示范:顺序搞反了
@retry(max_attempts=3)
@cache
def fetch_data(url):
pass
# ✅ 正确示范:先检查缓存,再考虑重试
@cache
@retry(max_attempts=3)
def fetch_data(url):
pass
解释:装饰器从下往上执行,所以最近的@cache先运行。如果把@retry放外层,缓存就不生效了——每次重试都会「缓存未命中」。
坑 2:忘记处理返回值
# ❌ 错误示范:装饰器没有 return 原函数的返回值
def bad_decorator(func):
def wrapper(*args, **kwargs):
print("执行前")
func(*args, **kwargs) # 没有 return!
print("执行后")
return wrapper
@bad_decorator
def add(a, b):
return a + b
result = add(1, 2)
print(result) # 输出:None(丢掉了返回值!)
# ✅ 正确示范:return result
def good_decorator(func):
def wrapper(*args, **kwargs):
print("执行前")
result = func(*args, **kwargs)
print("执行后")
return result # 重要!
return wrapper
坑 3:functools.wraps 忘了写
# ❌ 错误示范:没写 functools.wraps
def decorator(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper # 少了 @functools.wraps(func)
# ✅ 正确示范
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
后果:被装饰的函数会「失忆」,__name__变成wrapper,__doc__变成None,调试时很痛苦。
坑 4:装饰器带参数时写错位置
# ❌ 错误示范:@的位置不对
@timer # 错误:这个括号里不能放参数
def my_func():
pass
# ✅ 正确示范
@timer() # 如果timer需要参数
def my_func():
pass
# 或者不需要参数的装饰器
@simple_decorator # 直接用
def my_func():
pass
坑 5:类装饰器忘记保存原函数
# ❌ 错误示范
class CountCalls:
def __init__(self, func):
self.func = func # 保存了
def __call__(self, *args, **kwargs):
self.count += 1 # 但没有初始化 self.count!
return self.func(*args, **kwargs)
# ✅ 正确示范:初始化所有需要的属性
class CountCalls:
def __init__(self, func):
self._func = func # 用 _func 避免命名冲突
self.count = 0
functools.update_wrapper(self, func)
调试技巧:用 print 打日志
装饰器里最简单直接的调试方式就是print:
def debug(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"[DEBUG] 调用 {func.__name__},参数={args}, {kwargs}")
result = func(*args, **kwargs)
print(f"[DEBUG] 返回 {result}")
return result
return wrapper
如果想更专业一点,用logging模块:
import logging
def logged(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
logging.info(f"调用 {func.__name__}")
return func(*args, **kwargs)
return wrapper
✏️ 练习题 + 作业题
练习题(10 分钟)
练习 1(2 分钟):抄改计时器
- 输入:把项目1的计时器装饰器用到任意函数上
- 预期输出:显示函数执行时间
- 提示:复制@timer装饰器和slow_function,用自己的函数测试
练习 2(2 分钟):加一个条件判断
- 输入:在项目1的timer装饰器里,加一个判断,只在执行时间超过0.1秒时才打印
- 预期输出:快函数不打印,慢函数才打印
- 提示:在print之前加if end - start > 0.1
练习 3(3 分钟):改造成日志装饰器
- 输入:用项目2的@logger,记录一个计算函数的所有调用
- 预期输出:每次调用都打印日志,包含函数名和参数
- 提示:直接@logger放在你的函数上面
练习 4(5 分钟):组合两个装饰器
- 输入:给项目3的fetch_data加一个@logger装饰器
- 预期输出:既显示缓存命中/未命中,又显示完整调用日志
- 提示:注意装饰器叠加顺序
练习 5(3 分钟):找错
- 输入:下面这段代码为什么result是None?
def bad_cache(func):
def wrapper(*args, **kwargs):
if not hasattr(wrapper, '_cache'):
wrapper._cache = {}
key = str(args)
if key in wrapper._cache:
return wrapper._cache[key]
result = func(*args, **kwargs)
# 忘了保存到缓存
return result
return wrapper
- 预期输出:解释原因并给出修复方法
- 提示:检查缓存的保存逻辑
作业题(30 分钟 - 2 小时)
做一个「第9章 9.2 装饰器实战工具」—— 带重试和熔断的 API 调用器
-
需求描述:写一个 Python 脚本,模拟调用外部 API,用装饰器实现「调用失败自动重试」「连续失败N次后熔断」「记录每次调用」三个功能。
-
功能点:
1.@retry(次数, 间隔):失败自动重试,间隔N秒
2.@circuit_breaker(失败次数阈值, 熔断恢复时间):连续失败超过阈值,熔断一段时间
3.@api_logger:记录每次调用(时间、URL、参数、成功/失败) -
加分项:
1. 熔断时调用直接返回,不真正请求
2. 用装饰器参数控制是否启用功能 -
验收标准:
- 能跑起来,不报错
- 模拟失败时能看到重试日志
- 连续失败后能看到熔断日志
-
每次调用都有完整日志
-
提交方式:评论区贴代码或 GitHub 链接
📚 总结 + 资源
一句话总结:装饰器是 Python 里「不修改原函数就能增强功能」的利器,核心是闭包,关键要记住@functools.wraps保持身份、*args/**kwargs传递参数。
延伸学习:
1. 官方文档 - functools模块:https://docs.python.org/zh-cn/3/library/functools.html
2. 《Python编程:从入门到实践》第9章「装饰器与元类」
3. Real Python 教程 - Python Decorators:https://realpython.com/decorators/
互动钩子:你在项目里用过装饰器吗?遇到过什么坑?评论区聊聊,帮你答疑!
📌 下章预告:学会了装饰器,下一章我们要用它解决一个新问题——文件操作时「打开了就必须关上」的繁琐。下一章「9.3 上下文管理器 with 进阶」,教你用一行
with语句优雅管理资源,打开文件、连接数据库再也不怕忘关了!

评论(0)