Python 装饰器：从原理到高级应用

Python 装饰器是一种强大而优雅的代码复用机制，它允许我们在不修改原函数代码的情况下，为函数添加额外的功能。本文将从装饰器的基本原理开始，逐步深入到高级应用场景，包括缓存、日志、权限验证、重试机制等实战案例。

一、装饰器的工作原理

装饰器本质上是一个高阶函数，它接收一个函数作为参数，并返回一个新的函数。Python 使用 @ 语法糖来让装饰器的使用更加简洁和直观。

让我们从一个简单的例子开始，理解装饰器的基本结构：

def measure_time(func):
    """
    测量函数执行时间的装饰器
    """
    import time
    
    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 result
    
    return wrapper

# 使用装饰器函数
def slow_function():
    """模拟一个耗时操作"""
    import time
    time.sleep(1)
    return "完成"

# 使用 @ 语法糖
@measure_time
def fast_function():
    """模拟一个快速操作"""
    import time
    time.sleep(0.1)
    return "完成"

在这个例子中，measure_time 是一个装饰器，它接收一个函数 func，返回一个 wrapper 函数。wrapper 函数在执行 func 的前后添加了计时逻辑。当我们使用 @measure_time 装饰一个函数时，Python 会自动将函数传递给装饰器，并将装饰器返回的新函数赋值给原函数名。

二、保留函数元信息

使用装饰器时，原函数的元信息（如 __name__、__doc__）会被 wrapper 函数覆盖。为了解决这个问题，我们可以使用 functools.wraps 装饰器：

import functools
import time

def measure_time(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 result
    
    return wrapper

@measure_time
def complex_calculation(n):
    """
    计算斐波那契数列第 n 项
    
    Args:
        n: 项数
    
    Returns:
        斐波那契数列第 n 项的值
    """
    if n <= 1:
        return n
    return complex_calculation(n-1) + complex_calculation(n-2)

print(complex_calculation.__name__)  # 输出: complex_calculation
print(complex_calculation.__doc__)   # 输出函数的文档字符串

使用 functools.wraps 装饰器可以确保 wrapper 函数保持原函数的元信息，这对于调试和文档生成非常重要。

三、带参数的装饰器

有时候我们需要创建带参数的装饰器。这需要使用三层嵌套函数：最外层接收装饰器参数，中间层接收被装饰的函数，最内层是实际的包装函数。

import functools
import time

def repeat(times=1, delay=0):
    """
    重复执行函数的装饰器
    
    Args:
        times: 重复次数
        delay: 每次执行的延迟时间（秒）
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for i in range(times):
                if i > 0 and delay > 0:
                    time.sleep(delay)
                result = func(*args, **kwargs)
                results.append(result)
                print(f"第 {i+1} 次执行完成")
            return results
        
        return wrapper
    return decorator

@repeat(times=3, delay=1)
def fetch_data(url):
    """模拟从 URL 获取数据"""
    print(f"正在从 {url} 获取数据...")
    return f"数据来自 {url}"

results = fetch_data("https://api.example.com/data")
print(f"结果: {results}")

这个 repeat 装饰器可以指定重复执行的次数和每次执行之间的延迟时间，非常适合用于网络请求的重试场景。

四、实战应用一：缓存装饰器

缓存是装饰器的一个重要应用场景。我们可以创建一个缓存装饰器，缓存函数的结果，避免重复计算。

import functools

def cache(maxsize=None):
    """
    缓存装饰器，缓存函数的调用结果
    
    Args:
        maxsize: 最大缓存条目数，None 表示无限制
    """
    def decorator(func):
        cache_dict = {}
        call_count = [0]  # 使用列表以便在闭包中修改
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key = (args, frozenset(kwargs.items()))
            
            # 检查缓存
            if key in cache_dict:
                print(f"缓存命中: {func.__name__}{args}")
                return cache_dict[key]
            
            # 计算结果并缓存
            call_count[0] += 1
            result = func(*args, **kwargs)
            
            # 检查缓存大小限制
            if maxsize is not None and len(cache_dict) >= maxsize:
                # 简单的 FIFO 缓存淘汰策略
                oldest_key = next(iter(cache_dict))
                del cache_dict[oldest_key]
                print(f"缓存已满，淘汰: {oldest_key}")
            
            cache_dict[key] = result
            print(f"缓存未命中，实际调用: {func.__name__}{args}")
            return result
        
        # 添加缓存管理方法
        def cache_clear():
            """清除缓存"""
            cache_dict.clear()
            call_count[0] = 0
            print("缓存已清除")
        
        def cache_info():
            """获取缓存信息"""
            return {
                'hits': call_count[0] - len(cache_dict),
                'misses': len(cache_dict),
                'size': len(cache_dict)
            }
        
        wrapper.cache_clear = cache_clear
        wrapper.cache_info = cache_info
        return wrapper
    
    return decorator

@cache(maxsize=10)
def fibonacci(n):
    """计算斐波那契数列第 n 项（带缓存）"""
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 测试缓存效果
print(f"fibonacci(35) = {fibonacci(35)}")
print(f"fibonacci(35) = {fibonacci(35)}")  # 这次会从缓存读取
print(f"缓存信息: {fibonacci.cache_info()}")
fibonacci.cache_clear()

这个缓存装饰器实现了基本的缓存功能，包括缓存命中率统计和缓存大小限制。对于计算时间长的函数，使用缓存可以显著提高性能。

五、实战应用二：日志装饰器

日志记录是另一个装饰器的典型应用场景。我们可以创建一个灵活的日志装饰器，记录函数的调用参数、返回值和执行时间。

import functools
import time
import logging

def log(level=logging.INFO, show_args=True, show_result=True, show_time=True):
    """
    日志装饰器
    
    Args:
        level: 日志级别
        show_args: 是否显示调用参数
        show_result: 是否显示返回值
        show_time: 是否显示执行时间
    """
    def decorator(func):
        logger = logging.getLogger(func.__module__)
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 记录函数调用
            if show_args:
                args_str = ', '.join(repr(arg) for arg in args)
                kwargs_str = ', '.join(f'{k}={v!r}' for k, v in kwargs.items())
                params = ', '.join(filter(None, [args_str, kwargs_str]))
                logger.log(level, f"调用 {func.__name__}({params})")
            
            # 记录执行时间
            if show_time:
                start_time = time.time()
            
            # 执行函数
            try:
                result = func(*args, **kwargs)
                
                # 记录返回值
                if show_result:
                    logger.log(level, f"{func.__name__} 返回: {result!r}")
                
                return result
            
            except Exception as e:
                # 记录异常
                logger.error(f"{func.__name__} 抛出异常: {e}")
                raise
            
            finally:
                # 记录执行时间
                if show_time:
                    end_time = time.time()
                    logger.log(level, f"{func.__name__} 执行耗时: {end_time - start_time:.4f} 秒")
        
        return wrapper
    return decorator

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

@log(level=logging.INFO)
def divide(a, b):
    """除法运算"""
    return a / b

@log(level=logging.DEBUG, show_args=False, show_result=False)
def process_data(data):
    """处理数据"""
    import time
    time.sleep(0.1)
    return [x * 2 for x in data]

# 测试日志装饰器
print(divide(10, 2))
print(process_data([1, 2, 3, 4, 5]))

try:
    divide(10, 0)  # 这会触发异常
except ZeroDivisionError:
    print("捕获到除零异常")

这个日志装饰器提供了灵活的配置选项，可以根据需要记录函数调用的不同方面。它还正确处理了异常情况，确保异常信息被记录到日志中。

六、实战应用三：权限验证装饰器

在 Web 应用和 API 开发中，权限验证是一个常见需求。装饰器可以很优雅地实现权限检查。

import functools

# 模拟用户数据库
USERS = {
    'admin': {'role': 'admin', 'permissions': ['read', 'write', 'delete']},
    'user': {'role': 'user', 'permissions': ['read', 'write']},
    'guest': {'role': 'guest', 'permissions': ['read']}
}

class UnauthorizedError(Exception):
    """未授权异常"""
    pass

def require_role(*roles):
    """
    要求特定角色的装饰器
    
    Args:
        roles: 允许的角色列表
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 假设第一个参数是当前用户名
            username = args[0]
            
            if username not in USERS:
                raise UnauthorizedError(f"用户 {username} 不存在")
            
            user = USERS[username]
            if user['role'] not in roles:
                raise UnauthorizedError(f"用户 {username} 需要 {roles} 角色才能访问")
            
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

def require_permission(*permissions):
    """
    要求特定权限的装饰器
    
    Args:
        permissions: 需要的权限列表
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            username = args[0]
            
            if username not in USERS:
                raise UnauthorizedError(f"用户 {username} 不存在")
            
            user = USERS[username]
            user_permissions = user['permissions']
            
            # 检查是否拥有所有需要的权限
            missing = set(permissions) - set(user_permissions)
            if missing:
                raise UnauthorizedError(f"用户 {username} 缺少权限: {missing}")
            
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

# 使用权限装饰器
@require_role('admin')
def delete_user(operator, target_user):
    """删除用户（仅管理员）"""
    print(f"管理员 {operator} 删除用户 {target_user}")
    return f"用户 {target_user} 已删除"

@require_permission('write')
def update_profile(username, data):
    """更新用户资料（需要写权限）"""
    print(f"用户 {username} 更新资料: {data}")
    return "资料更新成功"

# 测试权限验证
print(delete_user('admin', 'test_user'))  # 成功

try:
    print(delete_user('user', 'test_user'))  # 失败，角色不符
except UnauthorizedError as e:
    print(f"错误: {e}")

print(update_profile('user', {'name': 'New Name'}))  # 成功

try:
    print(update_profile('guest', {'name': 'New Name'}))  # 失败，权限不足
except UnauthorizedError as e:
    print(f"错误: {e}")

这个权限验证装饰器系统提供了基于角色和基于权限的两种验证方式。它们可以组合使用，也可以单独使用，非常灵活。

七、实战应用四：重试机制装饰器

在网络请求和 IO 操作中，失败是常见的情况。使用重试装饰器可以自动重试失败的函数调用，提高系统的健壮性。

import functools
import time

def retry(max_attempts=3, delay=1, backoff=2, exceptions=(Exception,)):
    """
    重试装饰器
    
    Args:
        max_attempts: 最大尝试次数
        delay: 初始延迟时间（秒）
        backoff: 延迟时间的指数增长因子
        exceptions: 需要重试的异常类型
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            current_delay = delay
            
            for attempt in range(1, max_attempts + 1):
                try:
                    if attempt > 1:
                        print(f"第 {attempt} 次尝试...")
                    return func(*args, **kwargs)
                
                except exceptions as e:
                    last_exception = e
                    print(f"尝试 {attempt} 失败: {e}")
                    
                    if attempt < max_attempts:
                        print(f"等待 {current_delay:.2f} 秒后重试...")
                        time.sleep(current_delay)
                        current_delay *= backoff
            
            # 所有尝试都失败，抛出最后一个异常
            raise last_exception
        
        return wrapper
    return decorator

# 模拟不稳定的网络服务
class NetworkError(Exception):
    """网络错误"""
    pass

call_count = [0]

@retry(max_attempts=5, delay=1, backoff=2, exceptions=(NetworkError,))
def fetch_data_from_server():
    """模拟从服务器获取数据（可能失败）"""
    call_count[0] += 1
    print(f"尝试连接服务器...")
    
    # 前 3 次失败，第 4 次成功
    if call_count[0] < 4:
        raise NetworkError("连接超时")
    
    return {"status": "success", "data": "重要数据"}

# 测试重试机制
try:
    result = fetch_data_from_server()
    print(f"成功获取数据: {result}")
    print(f"总尝试次数: {call_count[0]}")
except NetworkError as e:
    print(f"所有尝试失败: {e}")

这个重试装饰器实现了指数退避算法，每次失败后等待的时间会按指数增长。这可以避免在服务器压力大时持续重试导致的问题。

八、装饰器类

除了使用函数作为装饰器，我们还可以使用类来实现装饰器。类装饰器可以更好地管理状态，并且可以更方便地添加额外的方法。

import functools
import time

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} 次调用 {self.func.__name__}")
        return self.func(*args, **kwargs)
    
    def reset(self):
        """重置计数器"""
        self.count = 0
    
    def get_count(self):
        """获取调用次数"""
        return self.count

# 使用装饰器类
@CountCalls
def process_item(item):
    """处理项目"""
    print(f"处理项目: {item}")
    return item.upper()

# 测试
process_item("hello")
process_item("world")
process_item("python")

print(f"总调用次数: {process_item.get_count()}")
process_item.reset()
print(f"重置后调用次数: {process_item.get_count()}")

类装饰器使用 __call__ 方法来实现装饰器的核心功能，并且可以添加额外的管理方法，如 reset() 和 get_count()。

九、多个装饰器叠加

Python 允许将多个装饰器叠加在一个函数上，装饰器的执行顺序是从下到上（从内到外）。

import functools

def bold(func):
    """添加粗体标记"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

def italic(func):
    """添加斜体标记"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

def underline(func):
    """添加下划线标记"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

# 多个装饰器叠加
@bold
@italic
@underline
def format_text(text):
    """格式化文本"""
    return text

# 测试
result = format_text("Hello, Python!")
print(result)  # 输出: Hello, Python!

在这个例子中，format_text 首先被 underline 装饰，然后被 italic 装饰，最后被 bold 装饰。装饰器的执行顺序是从下到上。

十、最佳实践指南

在使用装饰器时，应该遵循以下最佳实践：

1. 总是使用 functools.wraps：这可以确保被装饰函数的元信息被保留，这对于调试和文档生成非常重要。

2. 保持装饰器简单：装饰器应该专注于单一功能，不要在装饰器中实现过于复杂的逻辑。

3. 文档化装饰器：为装饰器编写清晰的文档字符串，说明装饰器的用途、参数和效果。

4. 考虑性能：装饰器会增加函数调用的开销，对于性能敏感的代码，要谨慎使用装饰器。

5. 正确处理异常：装饰器应该正确处理和传递异常，不要在装饰器中吞掉异常。

6. 避免可变默认参数：装饰器中要避免使用可变的默认参数，这可能导致意外的共享状态。

十一、总结

Python 装饰器是一种强大而优雅的代码复用机制，它可以在不修改原函数代码的情况下，为函数添加额外的功能。本文从装饰器的基本原理开始，介绍了如何创建简单装饰器、带参数的装饰器、装饰器类以及多个装饰器的叠加。

我们通过多个实战案例展示了装饰器的实际应用场景，包括缓存、日志、权限验证和重试机制等。这些案例都是实际开发中常见的需求，使用装饰器可以让代码更加简洁、可维护和可复用。

掌握装饰器是成为高级 Python 程序员的重要一步。合理使用装饰器可以显著提高代码质量和开发效率。希望本文能够帮助你更好地理解和应用 Python 装饰器。