一、为什么我们需要装饰器?

在Python开发中,你是否遇到过这样的场景:需要在多个函数执行前后添加日志记录?或者需要统计多个函数的执行时间?又或者需要对某些函数添加权限校验?

如果采用传统方式,我们可能需要在每个函数内部重复编写这些代码:

def func1():
    print("函数开始执行...")  # 重复代码
    # 核心业务逻辑
    print("函数执行完毕")     # 重复代码

def func2():
    print("函数开始执行...")  # 又一次重复
    # 核心业务逻辑
    print("函数执行完毕")     # 又一次重复

这种方式存在明显的问题:代码冗余、难以维护、修改时容易遗漏。这正是装饰器大显身手的场景。装饰器提供了一种优雅的方式,在不修改原函数代码的前提下,为函数添加额外的功能。


二、什么是装饰器?核心概念解析

2.1 装饰器的本质

在Python中,装饰器本质上是一个函数,它接收一个函数作为参数,并返回一个新的函数。听起来有点抽象?让我们一步步拆解。

关键概念一:Python中函数是一等公民

这意味着函数可以像普通变量一样被传递、赋值和返回。请看下面这个简单的例子:

# 定义一个普通函数
def say_hello():
    """这是一个打招呼的函数"""
    return "你好,世界!"

# 将函数赋值给一个变量
# 注意:say_hello后面没有括号,表示我们引用的是函数对象本身
greeting = say_hello

# 通过变量调用函数
result = greeting()
print(result)  # 输出:你好,世界!

# 这说明函数可以作为值进行传递

上面的代码展示了Python的一个核心特性:函数名本质上是一个指向函数对象的引用。当我们把say_hello赋值给greeting时,两个变量都指向同一个函数对象。

关键概念二:函数可以作为参数传递

def execute_function(func):
    """
    这是一个高阶函数,它接收另一个函数作为参数
    
    参数:
        func: 一个可调用的函数对象
    
    这个设计模式非常灵活:
    - 我们可以在执行前后添加日志
    - 可以捕获异常
    - 可以添加计时功能
    """
    print(f"即将执行函数: {func.__name__}")
    
    # 调用传入的函数
    result = func()
    
    print(f"函数执行完毕")
    return result

# 定义一个将被传递的函数
def my_task():
    """这是实际要执行的任务"""
    return "任务完成!"

# 将my_task作为参数传递给execute_function
# 注意:这里传入的是函数对象,而不是函数调用的结果
final_result = execute_function(my_task)
print(final_result)

执行结果:

即将执行函数: my_task
函数执行完毕
任务完成!

这个例子展示了装饰器的核心思想:通过一个"包装函数"来增强另一个函数的功能。

关键概念三:函数可以作为返回值

def create_multiplier(factor):
    """
    这是一个"函数工厂",它根据参数创建并返回新的函数
    
    这种技术称为"闭包"(Closure),是装饰器的重要基础
    
    参数:
        factor: 乘数因子,决定了返回函数的行为
    
    返回:
        一个新的函数,该函数会将输入值乘以factor
    """
    # 在内部定义一个新函数
    def multiplier(number):
        # 这个内部函数"记住"了外部函数的变量factor
        # 这就是闭包的特性
        return number * factor
    
    # 返回内部函数(注意没有括号,返回的是函数对象)
    return multiplier

# 使用函数工厂创建特定的乘法函数
double = create_multiplier(2)      # 创建一个"乘以2"的函数
triple = create_multiplier(3)      # 创建一个"乘以3"的函数

# 使用返回的函数
print(double(5))   # 输出: 10  (5 * 2)
print(triple(5))   # 输出: 15  (5 * 3)

这个例子非常重要,它展示了闭包的概念。闭包是指内部函数记住了外部函数的变量,即使外部函数已经执行完毕。装饰器正是利用闭包来实现功能的增强。


2.2 手动实现一个装饰器

让我们把上面的三个概念组合起来,手动实现一个装饰器:

def my_decorator(original_function):
    """
    这是一个装饰器函数
    
    它接收一个原函数,返回一个增强后的新函数
    
    设计思路:
    1. 创建一个新函数(wrapper),在其中调用原函数
    2. 在调用前后添加额外的功能
    3. 返回这个新函数
    
    参数:
        original_function: 要被装饰的原函数
    """
    
    def wrapper():
        """包装函数:在原函数执行前后添加额外功能"""
        # === 前置增强 ===
        print("=" * 50)
        print(f"开始执行函数: {original_function.__name__}")
        
        # === 执行原函数 ===
        result = original_function()
        
        # === 后置增强 ===
        print(f"函数执行完毕: {original_function.__name__}")
        print("=" * 50)
        
        # 返回原函数的结果,保持行为一致
        return result
    
    # 返回包装函数(这是装饰器的关键!)
    return wrapper

# 定义一个需要被装饰的函数
def say_hello():
    """简单的打招呼函数"""
    print("你好,我是Python学习者!")
    return "问候已发送"

# 手动应用装饰器
# 装饰器的本质就是:new_func = decorator(old_func)
decorated_function = my_decorator(say_hello)

# 调用装饰后的函数
decorated_function()

执行结果:

==================================================
开始执行函数: say_hello
你好,我是Python学习者!
函数执行完毕: say_hello
==================================================

2.3 使用@语法糖

Python提供了@符号作为装饰器的语法糖,让代码更加简洁优雅:

# 定义装饰器
def my_decorator(func):
    def wrapper():
        print(f"开始执行: {func.__name__}")
        result = func()
        print(f"执行完毕: {func.__name__}")
        return result
    return wrapper

# 使用@语法糖应用装饰器
# 这两行代码等价于:say_hello = my_decorator(say_hello)
@my_decorator
def say_hello():
    """打招呼函数"""
    print("你好!")
    return "完成"

# 直接调用,装饰器会自动生效
say_hello()

@my_decorator这种写法是Python的语法糖,它完全等同于say_hello = my_decorator(say_hello)。这种写法更直观、更优雅,也是实际开发中最常用的方式。


三、装饰器的高级用法

3.1 处理带参数的函数

上面的装饰器有一个问题:它只能装饰没有参数的函数。如果原函数有参数怎么办?

import functools

def smart_decorator(func):
    """
    智能装饰器:可以处理任意参数的函数
    
    使用*args和**kwargs来接收任意数量和形式的参数:
    - *args: 接收任意数量的位置参数,组成一个元组
    - **kwargs: 接收任意数量的关键字参数,组成一个字典
    
    functools.wraps的作用:
    - 保留原函数的元信息(如__name__, __doc__等)
    - 如果不加这行,被装饰函数的名称会变成"wrapper"
    """
    @functools.wraps(func)  # 这一行非常重要!
    def wrapper(*args, **kwargs):
        """
        包装函数
        
        参数说明:
        - *args: 接收所有位置参数
        - **kwargs: 接收所有关键字参数
        """
        print(f"调用函数: {func.__name__}")
        print(f"位置参数: {args}")
        print(f"关键字参数: {kwargs}")
        
        # 将所有参数原样传递给原函数
        result = func(*args, **kwargs)
        
        print(f"函数返回: {result}")
        return result
    
    return wrapper

# 应用装饰器
@smart_decorator
def greet(name, greeting="你好"):
    """
    问候函数
    
    参数:
        name: 被问候者的名字
        greeting: 问候语,默认为"你好"
    """
    message = f"{greeting}{name}!"
    print(message)
    return message

# 测试:传递不同的参数
greet("张三")                      # 使用默认问候语
greet("李四", greeting="早上好")   # 使用自定义问候语
greet("王五", "晚上好")            # 位置参数形式

代码解析:

  1. *args的作用:当我们调用greet("王五", "晚上好")时,args会是("王五", "晚上好")这个元组。

  2. **kwargs的作用:当我们调用greet("李四", greeting="早上好")时,kwargs会是{"greeting": "早上好"}这个字典。

  3. functools.wraps的重要性:试着把@functools.wraps(func)这行注释掉,然后打印greet.__name__,你会发现它变成了wrapper而不是greet。这会破坏函数的元信息,影响调试和文档生成。


3.2 带参数的装饰器

有时候,我们希望装饰器本身也能接收参数,比如设置日志级别、配置超时时间等。这需要再加一层函数嵌套:

import functools

def repeat(times=1):
    """
    带参数的装饰器:让函数重复执行指定次数
    
    这里的结构是三层函数嵌套:
    1. repeat(times): 接收装饰器参数,返回真正的装饰器
    2. decorator(func): 接收被装饰的函数
    3. wrapper(*args, **kwargs): 包装函数,实现具体功能
    
    参数:
        times: 重复执行的次数,默认为1
    
    使用示例:
        @repeat(times=3)
        def my_func():
            pass
    """
    # 第一层:接收装饰器参数
    def decorator(func):
        # 第二层:接收被装饰的函数
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 第三层:实现装饰功能
            results = []
            for i in range(times):
                print(f"第 {i+1} 次执行")
                result = func(*args, **kwargs)
                results.append(result)
            return results
        return wrapper
    return decorator

# 使用示例
@repeat(times=3)
def say_hello(name):
    """打招呼函数"""
    print(f"你好,{name}!")
    return f"已问候{name}"

# 调用函数
results = say_hello("小明")
print(f"所有结果: {results}")

执行结果:

第 1 次执行
你好,小明!
第 2 次执行
你好,小明!
第 3 次执行
你好,小明!
所有结果: ['已问候小明', '已问候小明', '已问候小明']

工作原理:

当Python解释器看到@repeat(times=3)时,它会:

  1. 首先执行repeat(times=3),返回decorator函数
  2. 然后执行decorator(say_hello),返回wrapper函数
  3. 最后say_hello指向wrapper函数

所以整个过程可以理解为:

say_hello = repeat(times=3)(say_hello)

3.3 类装饰器

除了函数形式的装饰器,我们还可以用类来实现装饰器。类装饰器在某些场景下更加灵活,因为类可以保存状态:

class CountCalls:
    """
    类装饰器:统计函数被调用的次数
    
    原理:
    1. __init__方法在被装饰函数定义时调用,保存原函数引用
    2. __call__方法使实例变成可调用对象,在每次调用时执行
    
    类装饰器的优势:
    - 可以方便地保存状态(如调用次数)
    - 可以提供更多的方法来查询或重置状态
    - 代码结构更清晰
    """
    
    def __init__(self, func):
        """
        初始化方法
        
        参数:
            func: 被装饰的函数
        
        这个方法在装饰器应用时自动调用
        """
        self.func = func           # 保存原函数引用
        self.count = 0             # 初始化调用计数器
        functools.update_wrapper(self, func)  # 保留原函数的元信息
    
    def __call__(self, *args, **kwargs):
        """
        使实例变成可调用对象
        
        当我们调用被装饰的函数时,实际上调用的是这个方法
        
        参数:
            *args: 位置参数
            **kwargs: 关键字参数
        """
        self.count += 1            # 每次调用,计数器加1
        print(f"第 {self.count} 次调用 {self.func.__name__}")
        
        # 调用原函数并返回结果
        return self.func(*args, **kwargs)
    
    def reset_count(self):
        """重置计数器"""
        self.count = 0
        print("计数器已重置")
    
    def get_count(self):
        """获取当前调用次数"""
        return self.count

# 使用类装饰器
@CountCalls
def process_data(data):
    """数据处理函数"""
    print(f"正在处理数据: {data}")
    return f"处理完成: {data}"

# 多次调用函数
process_data("数据A")
process_data("数据B")
process_data("数据C")

# 查询调用次数
print(f"总调用次数: {process_data.get_count()}")

# 重置计数器
process_data.reset_count()
process_data("新数据")
print(f"重置后调用次数: {process_data.get_count()}")

执行结果:

第 1 次调用 process_data
正在处理数据: 数据A
第 2 次调用 process_data
正在处理数据: 数据B
第 3 次调用 process_data
正在处理数据: 数据C
总调用次数: 3
计数器已重置
第 1 次调用 process_data
正在处理数据: 新数据
重置后调用次数: 1

四、实战应用场景

4.1 计时装饰器

在性能优化中,我们经常需要测量函数的执行时间。使用装饰器可以优雅地实现这个功能:

import functools
import time

def timer(func):
    """
    计时装饰器:测量函数执行时间
    
    功能:
    1. 在函数执行前记录开始时间
    2. 在函数执行后计算耗时
    3. 打印耗时信息
    
    适用场景:
    - 性能分析
    - 找出慢查询
    - 监控API响应时间
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 记录开始时间
        start_time = time.perf_counter()  # perf_counter比time.time更精确
        
        # 执行原函数
        result = func(*args, **kwargs)
        
        # 计算耗时
        end_time = time.perf_counter()
        elapsed = end_time - start_time
        
        # 打印耗时信息(转换为毫秒更直观)
        print(f"函数 {func.__name__} 执行耗时: {elapsed*1000:.2f} 毫秒")
        
        return result
    
    return wrapper

# 实际应用示例
@timer
def calculate_sum(n):
    """
    计算从1到n的和
    
    参数:
        n: 上限值
    
    这是一个模拟耗时操作的函数
    """
    total = 0
    for i in range(1, n + 1):
        total += i
    return total

@timer
def simulate_api_call():
    """模拟API调用"""
    time.sleep(0.5)  # 模拟网络延迟
    return {"status": "success", "data": "some data"}

# 测试
result1 = calculate_sum(1000000)
print(f"计算结果: {result1}\n")

result2 = simulate_api_call()
print(f"API响应: {result2}")

执行结果:

函数 calculate_sum 执行耗时: 45.23 毫秒
计算结果: 500000500000

函数 simulate_api_call 执行耗时: 502.18 毫秒
API响应: {'status': 'success', 'data': 'some data'}

4.2 日志记录装饰器

在生产环境中,良好的日志记录对于问题排查至关重要:

import functools
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),      # 输出到文件
        logging.StreamHandler()              # 输出到控制台
    ]
)

def log_calls(level='INFO'):
    """
    日志装饰器:记录函数调用信息
    
    功能:
    1. 记录函数调用时间、名称、参数
    2. 记录函数返回值
    3. 捕获并记录异常
    
    参数:
        level: 日志级别,可以是'DEBUG', 'INFO', 'WARNING', 'ERROR'
    
    设计思想:
    通过日志级别参数,我们可以控制哪些函数的日志需要详细记录
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取日志记录方法
            log_func = getattr(logging, level.lower())
            
            # 记录函数调用信息
            log_func(f"调用函数: {func.__name__}")
            log_func(f"位置参数: {args}")
            log_func(f"关键字参数: {kwargs}")
            
            try:
                # 执行函数
                result = func(*args, **kwargs)
                
                # 记录返回值
                log_func(f"函数返回: {result}")
                
                return result
                
            except Exception as e:
                # 记录异常信息
                logging.error(f"函数 {func.__name__} 发生异常: {str(e)}")
                raise  # 重新抛出异常,不影响原有逻辑
        
        return wrapper
    return decorator

# 使用示例
@log_calls(level='INFO')
def divide(a, b):
    """
    除法函数
    
    参数:
        a: 被除数
        b: 除数
    
    返回:
        除法结果
    """
    return a / b

@log_calls(level='DEBUG')
def calculate_average(numbers):
    """计算平均值"""
    return sum(numbers) / len(numbers)

# 正常调用
result1 = divide(10, 2)
print(f"10 / 2 = {result1}\n")

# 会产生异常的调用
try:
    result2 = divide(10, 0)
except ZeroDivisionError:
    print("捕获到除零异常\n")

# 另一个函数
avg = calculate_average([1, 2, 3, 4, 5])
print(f"平均值: {avg}")

日志输出示例:

2026-03-20 13:32:15,123 - INFO - 调用函数: divide
2026-03-20 13:32:15,124 - INFO - 位置参数: (10, 2)
2026-03-20 13:32:15,124 - INFO - 关键字参数: {}
2026-03-20 13:32:15,124 - INFO - 函数返回: 5.0

4.3 缓存装饰器

对于计算量大的函数,缓存结果可以显著提升性能:

import functools
import time

def memoize(ttl=None):
    """
    缓存装饰器:缓存函数的返回值
    
    功能:
    1. 根据参数缓存函数结果
    2. 相同参数直接返回缓存,避免重复计算
    3. 可选的缓存过期时间(TTL)
    
    参数:
        ttl: 缓存存活时间(秒),None表示永不过期
    
    原理:
    使用字典存储缓存,参数作为键,结果作为值
    
    适用场景:
    - 递归计算(如斐波那契数列)
    - API调用结果缓存
    - 数据库查询结果缓存
    """
    def decorator(func):
        # 缓存字典:存储 {参数: (结果, 时间戳)}
        cache = {}
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 创建缓存键(需要处理不可哈希的参数)
            # 将args和kwargs转换为可哈希的形式
            key = (args, tuple(sorted(kwargs.items())))
            
            # 检查缓存是否存在
            if key in cache:
                cached_result, cached_time = cache[key]
                
                # 检查是否过期
                if ttl is None or (time.time() - cached_time) < ttl:
                    print(f"从缓存返回: {func.__name__}{args}")
                    return cached_result
            
            # 缓存不存在或已过期,执行函数
            print(f"执行计算: {func.__name__}{args}")
            result = func(*args, **kwargs)
            
            # 存入缓存
            cache[key] = (result, time.time())
            
            return result
        
        # 添加清除缓存的方法
        wrapper.clear_cache = lambda: cache.clear()
        wrapper.cache_info = lambda: f"缓存大小: {len(cache)}"
        
        return wrapper
    
    return decorator

# 使用示例
@memoize(ttl=10)  # 缓存10秒
def fibonacci(n):
    """
    计算斐波那契数列的第n项
    
    参数:
        n: 项数(从0开始)
    
    这是一个经典的递归问题,不缓存的话效率极低
    因为会重复计算很多相同的子问题
    """
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

@memoize(ttl=5)  # 缓存5秒
def expensive_computation(x):
    """模拟耗时计算"""
    time.sleep(1)  # 模拟计算延迟
    return x * x * x

# 测试斐波那契数列
print("计算斐波那契数列:")
start = time.time()
result = fibonacci(30)  # 第一次计算
print(f"fibonacci(30) = {result}, 耗时: {time.time()-start:.2f}秒")

start = time.time()
result = fibonacci(30)  # 第二次从缓存获取
print(f"fibonacci(30) = {result}, 耗时: {time.time()-start:.2f}秒\n")

# 测试带TTL的缓存
print("测试缓存过期:")
result1 = expensive_computation(5)  # 计算
print(f"第一次: {result1}")

result2 = expensive_computation(5)  # 从缓存
print(f"第二次: {result2}")

print("等待5秒后...")
time.sleep(5)

result3 = expensive_computation(5)  # 缓存过期,重新计算
print(f"第三次: {result3}")

4.4 权限验证装饰器

在Web应用中,权限验证是一个常见需求:

import functools

# 模拟用户数据库
users_db = {
    "admin": {"role": "admin", "permissions": ["read", "write", "delete"]},
    "editor": {"role": "editor", "permissions": ["read", "write"]},
    "viewer": {"role": "viewer", "permissions": ["read"]}
}

# 模拟当前用户(实际应用中从session或token获取)
current_user = {"username": "editor"}

def require_permission(permission):
    """
    权限验证装饰器
    
    功能:
    1. 检查当前用户是否具有指定权限
    2. 有权限则执行函数,无权限则拒绝访问
    
    参数:
        permission: 需要的权限名称
    
    设计思路:
    这是一种AOP(面向切面编程)的思想
    将权限验证逻辑与业务逻辑分离
    
    实际应用场景:
    - Web API权限控制
    - 功能模块访问控制
    - 数据操作权限验证
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取当前用户信息
            username = current_user.get("username")
            
            if not username:
                raise PermissionError("用户未登录")
            
            # 查询用户权限
            user_info = users_db.get(username)
            if not user_info:
                raise PermissionError("用户不存在")
            
            user_permissions = user_info.get("permissions", [])
            
            # 检查权限
            if permission not in user_permissions:
                raise PermissionError(
                    f"用户 {username} 没有 {permission} 权限。"
                    f"当前权限: {user_permissions}"
                )
            
            # 权限验证通过,执行函数
            print(f"✓ 权限验证通过: {username} 拥有 {permission} 权限")
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

# 应用示例:不同的操作需要不同的权限
@require_permission("read")
def view_document(doc_id):
    """查看文档"""
    return f"文档内容 #{doc_id}"

@require_permission("write")
def edit_document(doc_id, content):
    """编辑文档"""
    return f"文档 #{doc_id} 已更新为: {content}"

@require_permission("delete")
def delete_document(doc_id):
    """删除文档"""
    return f"文档 #{doc_id} 已删除"

# 测试
print("当前用户: editor (权限: read, write)\n")

# 可以执行的操作
try:
    result = view_document(1)
    print(result)
except PermissionError as e:
    print(f"权限错误: {e}")

try:
    result = edit_document(1, "新内容")
    print(result)
except PermissionError as e:
    print(f"权限错误: {e}")

# 无法执行的操作
try:
    result = delete_document(1)
    print(result)
except PermissionError as e:
    print(f"权限错误: {e}")

执行结果:

当前用户: editor (权限: read, write)

✓ 权限验证通过: editor 拥有 read 权限
文档内容 #1
✓ 权限验证通过: editor 拥有 write 权限
文档 #1 已更新为: 新内容
权限错误: 用户 editor 没有 delete 权限。当前权限: ['read', 'write']

4.5 重试装饰器

在网络请求等场景中,自动重试可以提高系统的可靠性:

import functools
import time
import random

def retry(max_attempts=3, delay=1, backoff=2, exceptions=(Exception,)):
    """
    重试装饰器:当函数失败时自动重试
    
    参数:
        max_attempts: 最大尝试次数(包括首次)
        delay: 初始延迟时间(秒)
        backoff: 延迟时间的增长因子
        exceptions: 需要重试的异常类型元组
    
    工作原理:
    1. 首次执行函数
    2. 如果失败且未达到最大次数,等待一段时间后重试
    3. 每次重试,等待时间按backoff倍数增长
    4. 达到最大次数后仍失败,抛出异常
    
    使用场景:
    - 网络请求
    - 数据库连接
    - 外部API调用
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            last_exception = None
            
            for attempt in range(1, max_attempts + 1):
                try:
                    # 尝试执行函数
                    result = func(*args, **kwargs)
                    if attempt > 1:
                        print(f"✓ 第 {attempt} 次尝试成功")
                    return result
                    
                except exceptions as e:
                    last_exception = e
                    
                    if attempt == max_attempts:
                        print(f"✗ 已达到最大重试次数 {max_attempts}")
                        raise
                    
                    print(f"✗ 第 {attempt} 次尝试失败: {str(e)}")
                    print(f"  等待 {current_delay} 秒后重试...")
                    
                    time.sleep(current_delay)
                    current_delay *= backoff  # 延迟时间指数增长
            
            # 理论上不会执行到这里
            raise last_exception
        
        return wrapper
    return decorator

# 模拟一个不稳定的API调用
call_count = 0

@retry(max_attempts=5, delay=1, backoff=2, exceptions=(ConnectionError, TimeoutError))
def unstable_api_call():
    """
    模拟不稳定的API调用
    
    前3次会失败,第4次成功
    这模拟了真实世界中网络抖动的场景
    """
    global call_count
    call_count += 1
    
    # 模拟前3次失败
    if call_count < 4:
        raise ConnectionError(f"连接失败 (第{call_count}次)")
    
    # 第4次成功
    return {"status": "success", "data": "API响应数据", "attempts": call_count}

# 测试
print("测试重试装饰器:\n")
try:
    result = unstable_api_call()
    print(f"\n最终结果: {result}")
except Exception as e:
    print(f"最终失败: {e}")

执行结果:

测试重试装饰器:

✗ 第 1 次尝试失败: 连接失败 (第1次)
  等待 1 秒后重试...
✗ 第 2 次尝试失败: 连接失败 (第2次)
  等待 2 秒后重试...
✗ 第 3 次尝试失败: 连接失败 (第3次)
  等待 4 秒后重试...
✓ 第 4 次尝试成功

最终结果: {'status': 'success', 'data': 'API响应数据', 'attempts': 4}

五、装饰器组合与执行顺序

当一个函数被多个装饰器装饰时,执行顺序是怎样的?

import functools

def decorator_a(func):
    """装饰器A"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("A: 执行前")
        result = func(*args, **kwargs)
        print("A: 执行后")
        return result
    return wrapper

def decorator_b(func):
    """装饰器B"""
    @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("执行函数主体")

print("调用 my_function:")
my_function()

print("\n" + "="*30)
print("执行顺序解析:")
print("="*30)
print("""
@decorator_a
@decorator_b
def my_function():
    pass

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

执行流程:
1. decorator_b 先包装 my_function,返回 wrapper_b
2. decorator_a 再包装 wrapper_b,返回 wrapper_a
3. my_function 最终指向 wrapper_a

调用时:
- wrapper_a 开始执行(打印 "A: 执行前")
- wrapper_a 调用 wrapper_b
  - wrapper_b 开始执行(打印 "B: 执行前")
  - wrapper_b 调用原函数(打印 "执行函数主体")
  - wrapper_b 结束执行(打印 "B: 执行后")
- wrapper_a 结束执行(打印 "A: 执行后")

总结:
- 装饰器从下往上应用(离函数近的先应用)
- 执行时从外到内(最外层的装饰器先执行"前"部分)
- 返回时从内到外(最内层的装饰器先执行"后"部分)
""")

执行结果:

调用 my_function:
A: 执行前
B: 执行前
执行函数主体
B: 执行后
A: 执行后

六、标准库中的常用装饰器

Python标准库提供了很多实用的装饰器:

6.1 functools.lru_cache

LRU(Least Recently Used)缓存装饰器,自动缓存函数结果:

from functools import lru_cache
import time

@lru_cache(maxsize=128)
def fibonacci(n):
    """
    使用LRU缓存的斐波那契函数
    
    maxsize参数:
    - 缓存的最大条目数
    - 设为None表示无限制
    - 当缓存满时,淘汰最久未使用的条目
    
    优点:
    - 无需手动实现缓存逻辑
    - 自动管理缓存大小
    - 线程安全
    """
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# 测试性能
start = time.time()
result = fibonacci(100)
end = time.time()

print(f"fibonacci(100) = {result}")
print(f"计算耗时: {end - start:.6f} 秒")  # 应该非常快

# 查看缓存信息
print(f"\n缓存信息: {fibonacci.cache_info()}")
# cache_info返回:命中次数、未命中次数、缓存大小、最大缓存大小

# 清除缓存
fibonacci.cache_clear()
print("缓存已清除")

6.2 contextlib.contextmanager

用于创建上下文管理器的装饰器:

from contextlib import contextmanager

@contextmanager
def timer_context(name):
    """
    计时上下文管理器
    
    使用with语句自动管理资源
    yiled之前是进入时执行的代码
    yield之后是退出时执行的代码
    """
    print(f"开始: {name}")
    start = time.time()
    
    try:
        yield  # 这里可以返回值,通过as子句获取
    finally:
        elapsed = time.time() - start
        print(f"结束: {name}, 耗时: {elapsed:.2f}秒")

# 使用示例
with timer_context("数据处理"):
    # 在这个代码块中执行的操作会被计时
    time.sleep(1)
    print("正在处理数据...")

print("\n使用返回值示例:")

@contextmanager
def database_connection(host):
    """模拟数据库连接管理"""
    print(f"连接数据库: {host}")
    connection = {"host": host, "status": "connected"}
    
    try:
        yield connection  # 返回连接对象
    finally:
        print(f"关闭连接: {host}")
        connection["status"] = "closed"

# 使用
with database_connection("localhost") as conn:
    print(f"使用连接: {conn}")
    print(f"连接状态: {conn['status']}")

6.3 dataclasses.dataclass

简化类定义的装饰器:

from dataclasses import dataclass, field
from typing import List

@dataclass
class Person:
    """
    使用dataclass定义数据类
    
    自动生成:
    - __init__ 方法
    - __repr__ 方法
    - __eq__ 方法
    
    参数说明:
    - order=True: 生成比较方法(__lt__, __le__, __gt__, __ge__)
    - frozen=True: 使实例不可变
    """
    name: str                    # 必填字段
    age: int                     # 必填字段
    email: str = ""              # 默认值
    hobbies: List[str] = field(default_factory=list)  # 可变默认值

# 创建实例
person1 = Person("张三", 25)
person2 = Person("李四", 30, "lisi@example.com")
person3 = Person("王五", 25, hobbies=["阅读", "游泳"])

print(person1)  # 自动生成的__repr__
print(person2)
print(person3)

print(f"\nperson1 == person2: {person1 == person2}")  # 自动生成的__eq__

七、总结

7.1 装饰器的核心要点

  1. 本质理解:装饰器是一个接收函数并返回新函数的高阶函数,利用了Python的闭包特性。

  2. 语法糖@decorator语法等价于func = decorator(func),让代码更简洁。

  3. 参数处理:使用*args**kwargs处理任意参数,functools.wraps保留原函数元信息。

  4. 带参数装饰器:通过三层嵌套实现,最外层接收装饰器参数。

  5. 类装饰器:通过__call__方法实现,适合需要保存状态的场景。

7.2 最佳实践

  1. 始终使用functools.wraps:保留原函数的名称、文档字符串等信息。

  2. 装饰器应该有单一职责:一个装饰器只做一件事,便于组合和测试。

  3. 提供配置选项:通过参数让装饰器行为可配置。

  4. 正确处理异常:装饰器中的异常处理要透明,不要吞掉重要异常。

  5. 文档化装饰器:清楚地说明装饰器的功能、参数和使用方式。

7.3 常见应用场景

场景 装饰器示例 作用
日志记录 @log_calls 记录函数调用信息
性能测量 @timer 测量函数执行时间
权限控制 @require_permission 验证用户权限
缓存优化 @lru_cache 缓存函数结果
错误重试 @retry 自动重试失败的操作
参数验证 @validate_args 验证函数参数

装饰器是Python中最优雅的设计模式之一,它体现了"开放-封闭原则"——对扩展开放,对修改封闭。通过装饰器,我们可以在不修改原有代码的情况下,灵活地为函数添加新功能,这正是Python"Pythonic"哲学的完美体现。


作者注:本文系统性地介绍了Python装饰器的原理和实践,所有代码都经过精心设计,包含详细注释和执行结果说明。建议读者在学习时,将每段代码亲自运行一遍,加深理解。装饰器虽然概念简单,但应用广泛,掌握它将让你的Python代码更加优雅和高效。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐