第 6 讲：Python 装饰器完全指南 | @wraps、类装饰器、带参装饰器与 AOP 编程

大家好，我是正在实战各种 AI 项目的程序员晚枫。

🎬 开篇：装饰器的本质

你有没有见过这样的代码？

@timer
def slow_function():
    time.sleep(1)
    return "Done"

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

@retry(times=3)
def fetch_data():
    return requests.get("https://api.example.com")

这些 @xxx 就是装饰器，它们是 Python 最优雅的特性之一。

装饰器的本质是什么？

简单来说：装饰器是一个函数，它接受一个函数作为参数，返回一个新的函数。

# 装饰器本质
@decorator
def func():
    pass

# 等价于
func = decorator(func)

今天我们就彻底掌握装饰器的原理和应用。

---

🎨 基础装饰器

最简单的装饰器

def my_decorator(func):
    """最简单的装饰器"""
    def wrapper():
        print("调用前")
        func()  # 调用原函数
        print("调用后")
    return wrapper

@my_decorator
def say_hello():
    print("Hello!")

# 调用
say_hello()
# 输出:
# 调用前
# Hello!
# 调用后

# 等价于
# say_hello = my_decorator(say_hello)

处理参数和返回值

def my_decorator(func):
    """处理任意参数的装饰器"""
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}，参数: {args}, {kwargs}")
        result = func(*args, **kwargs)  # 调用原函数
        print(f"返回值: {result}")
        return result
    return wrapper

@my_decorator
def add(a, b):
    return a + b

@my_decorator
def greet(name, greeting="Hello"):
    return f"{greeting}, {name}!"

print(add(3, 5))
# 调用 add，参数: (3, 5), {}
# 返回值: 8
# 8

print(greet("Alice", greeting="Hi"))
# 调用 greet，参数: ('Alice',), {'greeting': 'Hi'}
# 返回值: Hi, Alice!
# Hi, Alice!

实用装饰器示例

1. 计时装饰器

import time
import functools

def timer(func):
    """测量函数执行时间"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} 执行时间: {elapsed:.6f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)
    return "Done"

slow_function()  # slow_function 执行时间: 1.001234s

2. 日志装饰器

import functools
from datetime import datetime

def log_calls(func):
    """记录函数调用日志"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{timestamp}] 调用 {func.__name__}")
        print(f"  参数: args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"  返回: {result}")
        return result
    return wrapper

@log_calls
def calculate(a, b, operation="add"):
    if operation == "add":
        return a + b
    elif operation == "multiply":
        return a * b

calculate(3, 5, operation="multiply")
# [2024-01-15 10:30:00] 调用 calculate
#   参数: args=(3, 5), kwargs={'operation': 'multiply'}
#   返回: 15

3. 缓存装饰器

import functools

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] = result
        return result
    
    # 提供清空缓存的方法
    wrapper.cache_clear = cache.clear
    return wrapper

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

print(fibonacci(10))  # 计算
print(fibonacci(10))  # 缓存命中
fibonacci.cache_clear()  # 清空缓存

---

🔧 functools.wraps：保留原函数信息

为什么需要 @wraps？

# ❌ 不使用 wraps
def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def my_function():
    """这是一个重要函数"""
    return "result"

print(my_function.__name__)  # wrapper（错误！）
print(my_function.__doc__)   # None（错误！）
print(my_function)           # <function bad_decorator.<locals>.wrapper at 0x...>

# ✅ 使用 wraps
import functools

def good_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@good_decorator
def my_function():
    """这是一个重要函数"""
    return "result"

print(my_function.__name__)  # my_function（正确！）
print(my_function.__doc__)   # 这是一个重要函数（正确！）
print(my_function)           # <function my_function at 0x...>

functools.wraps 做了什么？

# wraps 实际上复制了原函数的元信息
WRAPPER_ASSIGNMENTS = (
    '__module__', '__name__', '__qualname__', '__annotations__',
    '__doc__', '__wrapped__'
)
WRAPPER_UPDATES = ('__dict__',)

# 手动实现（不推荐，用 @wraps 即可）
def manual_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    wrapper.__name__ = func.__name__
    wrapper.__doc__ = func.__doc__
    wrapper.__wrapped__ = func
    return wrapper

---

🔧 参数化装饰器

带参数的装饰器

import functools

def repeat(n):
    """重复执行 n 次的装饰器工厂"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(n):
                result = func(*args, **kwargs)
                results.append(result)
            return results[-1] if results else None
        return wrapper
    return decorator

@repeat(3)
def greet(name):
    print(f"Hello, {name}!")
    return name

greet("Alice")
# Hello, Alice!
# Hello, Alice!
# Hello, Alice!

装饰器的三层结构

# 参数化装饰器有三层：
# 1. 外层：接收装饰器参数
# 2. 中层：接收被装饰的函数
# 3. 内层：实际执行的包装函数

def decorator_with_args(arg1, arg2):
    """三层装饰器示例"""
    print(f"装饰器参数: {arg1}, {arg2}")
    
    def decorator(func):
        print(f"装饰函数: {func.__name__}")
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print(f"执行前: {arg1}")
            result = func(*args, **kwargs)
            print(f"执行后: {arg2}")
            return result
        
        return wrapper
    
    return decorator

@decorator_with_args("before", "after")
def my_function():
    print("函数执行中")

my_function()

实用的参数化装饰器

1. 重试装饰器

import functools
import time

def retry(times=3, delay=1, exceptions=(Exception,)):
    """重试装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(times):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == times - 1:
                        raise
                    print(f"第 {attempt + 1} 次失败: {e}，{delay}秒后重试")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(times=3, delay=1, exceptions=(ConnectionError,))
def fetch_data():
    import random
    if random.random() < 0.7:  # 70% 概率失败
        raise ConnectionError("网络错误")
    return "数据"

# fetch_data()  # 会重试最多 3 次

2. 超时装饰器

import functools
import signal

def timeout(seconds):
    """超时装饰器（仅 Unix）"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            def handle_timeout(signum, frame):
                raise TimeoutError(f"函数 {func.__name__} 超时")
            
            signal.signal(signal.SIGALRM, handle_timeout)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result
        return wrapper
    return decorator

@timeout(5)
def slow_function():
    time.sleep(10)
    return "Done"

# slow_function()  # 5秒后抛出 TimeoutError

3. 权限检查装饰器

import functools

def require_role(*roles):
    """权限检查装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(user, *args, **kwargs):
            if user.get('role') not in roles:
                raise PermissionError(
                    f"需要角色: {roles}，当前角色: {user.get('role')}"
                )
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

@require_role('admin', 'manager')
def delete_user(user, user_id):
    print(f"用户 {user['name']} 删除了用户 {user_id}")
    return True

# 正确权限
admin = {'name': 'Alice', 'role': 'admin'}
delete_user(admin, 123)  # 成功

# 错误权限
user = {'name': 'Bob', 'role': 'user'}
# delete_user(user, 123)  # PermissionError

---

🏗️ 类装饰器

用类实现装饰器

import functools

class CountCalls:
    """统计调用次数的类装饰器"""
    
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
        self.num_calls = 0
    
    def __call__(self, *args, **kwargs):
        self.num_calls += 1
        print(f"第 {self.num_calls} 次调用 {self.func.__name__}")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello!")

say_hello()  # 第 1 次调用 say_hello
say_hello()  # 第 2 次调用 say_hello
say_hello()  # 第 3 次调用 say_hello

带状态的类装饰器

class RateLimiter:
    """速率限制装饰器"""
    
    def __init__(self, max_calls, period=60):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
    
    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import time
            now = time.time()
            
            # 清理过期记录
            self.calls = [t for t in self.calls if now - t < self.period]
            
            # 检查是否超限
            if len(self.calls) >= self.max_calls:
                raise RuntimeError(
                    f"调用超限: {self.max_calls}次/{self.period}秒"
                )
            
            self.calls.append(now)
            return func(*args, **kwargs)
        
        return wrapper

@RateLimiter(max_calls=5, period=60)
def api_call():
    return "API 响应"

# 5 次内正常
for _ in range(5):
    print(api_call())

# 第 6 次会抛出 RuntimeError
# api_call()  # RuntimeError

装饰器类 vs 函数装饰器

特性	函数装饰器	类装饰器
状态管理	需要闭包或 nonlocal	天然支持属性
可读性	简单场景更清晰	复杂逻辑更清晰
继承	不支持	支持
扩展性	有限	高

---

🔄 多个装饰器的堆叠

执行顺序

import functools

def decorator_a(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("A 开始")
        result = func(*args, **kwargs)
        print("A 结束")
        return result
    return wrapper

def decorator_b(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("B 开始")
        result = func(*args, **kwargs)
        print("B 结束")
        return result
    return wrapper

@decorator_a
@decorator_b
def my_function():
    print("函数执行")

my_function()
# A 开始
# B 开始
# 函数执行
# B 结束
# A 结束

# 等价于
# my_function = decorator_a(decorator_b(my_function))

规则：装饰器从下往上应用，执行从外往里。

实际应用

@timer
@retry(times=3)
@log_calls
def fetch_data(url):
    return requests.get(url).json()

# 执行顺序：
# 1. timer 开始计时
# 2. retry 准备重试
# 3. log_calls 记录日志
# 4. fetch_data 执行
# 5. log_calls 结束
# 6. retry 结束
# 7. timer 结束并打印时间

---

🎯 AOP（面向切面编程）

什么是 AOP？

AOP（Aspect-Oriented Programming）是一种编程范式，将横切关注点与业务逻辑分离。

横切关注点包括：

• 日志记录
• 性能监控
• 权限检查
• 事务管理
• 错误处理

用装饰器实现 AOP

import functools
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def aspect_log(func):
    """日志切面"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"调用 {func.__name__}")
        try:
            result = func(*args, **kwargs)
            logger.info(f"{func.__name__} 成功")
            return result
        except Exception as e:
            logger.error(f"{func.__name__} 失败: {e}")
            raise
    return wrapper

def aspect_timer(func):
    """性能切面"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        logger.info(f"{func.__name__} 耗时 {elapsed:.6f}s")
        return result
    return wrapper

def aspect_cache(ttl=60):
    """缓存切面"""
    def decorator(func):
        cache = {}
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import time
            key = (args, tuple(sorted(kwargs.items())))
            
            if key in cache:
                value, timestamp = cache[key]
                if time.time() - timestamp < ttl:
                    logger.info(f"{func.__name__} 缓存命中")
                    return value
            
            result = func(*args, **kwargs)
            cache[key] = (result, time.time())
            return result
        
        return wrapper
    return decorator

# 组合使用
@aspect_log
@aspect_timer
@aspect_cache(ttl=60)
def get_user_info(user_id):
    # 模拟耗时操作
    time.sleep(1)
    return {"id": user_id, "name": "Alice"}

get_user_info(1)  # 执行并缓存
get_user_info(1)  # 缓存命中

---

⚠️ 避坑指南

陷阱 1：忘记使用 @wraps

# ❌ 问题
def decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@decorator
def my_func(a, b):
    """求和"""
    return a + b

# 丢失了元信息
print(my_func.__name__)  # wrapper
print(my_func.__doc__)   # None

# ✅ 解决：使用 @functools.wraps

陷阱 2：装饰器影响调试

# 使用 __wrapped__ 获取原函数
@timer
def my_function():
    return 42

# 直接调用原函数
original = my_function.__wrapped__
print(original())  # 不经过装饰器

陷阱 3：类方法装饰器

# 类方法装饰器需要处理 self
def method_decorator(func):
    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        print(f"调用 {func.__name__}")
        return func(self, *args, **kwargs)
    return wrapper

class MyClass:
    @method_decorator
    def my_method(self):
        print("方法执行")

obj = MyClass()
obj.my_method()

陷阱 4：装饰器顺序错误

# ❌ 错误顺序
@staticmethod
@timer  # staticmethod 已经改变了函数类型，timer 无法正常工作
def my_static():
    pass

# ✅ 正确顺序
@timer
@staticmethod
def my_static():
    pass

---

🎯 本讲总结

通过本讲，我们掌握了：

知识点	核心要点
装饰器本质	接受函数，返回函数的高阶函数
@wraps	保留原函数的元信息
参数化装饰器	三层结构：工厂 → 装饰器 → 包装器
类装饰器	实现 __call__ 方法，便于管理状态
装饰器堆叠	从下往上应用，从外往里执行
AOP	用装饰器实现横切关注点分离

记住这句话：

装饰器是 Python 的语法糖，让你在不修改原函数的情况下，扩展函数的功能。

---

📚 推荐教材

《Python 编程从入门到实践（第 3 版）》 | 《流畅的 Python（第 2 版）》 | 《CPython 设计与实现》

学习路线： 零基础 → 《从入门到实践》 → 《流畅的 Python》 → 本门课程 → 《CPython 设计与实现》

---

🎓 加入《流畅的 Python》直播共读营

学到这里，如果你想系统吃透这本书——欢迎加入我的直播共读课。

• 每周直播精讲，逐章拆解核心知识点
• 专属学习群，随时答疑交流
• 试运营特惠：499 元 → 299 元

👉 【立即报名《流畅的 Python》共读课】：https://mp.weixin.qq.com/s/ivHJwn1nNx5ug4TFrapvGg

---

🔗 课程导航

← 上一讲：函数即对象 | 下一讲：生成器与协程 →

---

💬 联系我

平台	账号/链接
微信	扫码加好友
微博	@程序员晚枫
知乎	@程序员晚枫
抖音	@程序员晚枫
小红书	@程序员晚枫
B 站	Python 自动化办公社区

主营业务：AI 编程培训、企业内训、技术咨询

🎓 AI 编程实战课程

想系统学习 AI 编程？程序员晚枫的 AI 编程实战课 帮你从零上手！

• 👉 课程报名：点击这里报名，前3讲免费试听
• 👉 免费试看：B站免费试看前3讲，先看看适不适合自己