第9章 9.2 装饰器从原理到实战

🎯 开场:快递盒子的启示

上一章我们学会了列表推导式,用一条简洁的「点菜流水线」把繁琐的循环变成了优雅的一行代码。如果你已经能熟练用它来清洗数据,恭喜你——你已经从「会写代码」往「写好代码」迈了一大步。

但今天我要教你一个更神奇的工具,它能让你不动原代码就给函数加新功能。听起来很玄?听我讲个故事。

你有没有遇到过这种情况:写了一个函数,运行正常,但某天产品经理说「加上日志吧」「加上性能计时吧」「加上权限校验吧」——你一加,发现整个代码结构要改,或者好几个地方都要复制粘贴同样的代码。

痛点来了:想给函数加个通用功能,却要改函数本身,重复代码满天飞。

这就是今天要解决的——装饰器。学会了它,上面那些需求,一行代码搞定。


🧱 基础 25 分钟:核心概念

什么是装饰器?说白了就是「包装纸」

想象你有一个快递盒子(函数),你想在外面包一层气泡垫(额外功能),但又不想把盒子里的东西(函数本身)拆开重装。装饰器就是这个思路——在函数外面包一层包装,不动原函数就给它加功能

第一步:理解闭包——装饰器的「内功心法」

装饰器的底层原理是闭包。闭包是啥?

生活类比:你把一本书(变量)放在包里(函数),然后把包借给别人。别人虽然拿不到书,但下次他打开包的时候,还能看到那本书在里头。

用代码说话:

def 外层函数():
收藏的古董 = "1980年的邮票"  # 外部变量

def 内层函数():
    print(f"我看到了:{收藏的古董}")  # 引用外部变量
return 内层函数  # 把内层函数当成返回值

拿到的东西 = 外层函数()
拿到的东西()  # 输出:我看到了:1980年的邮票

内层函数记住了外层函数的变量,这就是闭包。闭包让内部函数能「记住」定义时的环境。

配图1 - 配图1

第二步:写第一个装饰器

为什么要用:不动原函数代码,给它加上新功能。

怎么用:定义一个「包装函数」,把被装饰的函数包进去。

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函数上面,顺序不能错。

配图2 - 配图2

第四步:带参数的装饰器——给包装纸加开口

如果被装饰的函数有参数怎么办?

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 分钟):找错
- 输入:下面这段代码为什么resultNone

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语句优雅管理资源,打开文件、连接数据库再也不怕忘关了!

声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理。