🐍 Python 核心技术

装饰器 Decorators

掌握 Python 中最优雅的代码复用模式,深入理解函数式编程的精髓

01

核心概念

📌 什么是装饰器?

装饰器(Decorator)是 Python 中一种强大的语法特性,它允许你在不修改原函数代码的情况下,动态地扩展函数的功能。装饰器本质上是一个接收函数作为参数并返回新函数的高阶函数。

装饰器使用 @ 符号作为语法糖,使代码更加简洁优雅。它是 Python 函数式编程和元编程的重要组成部分。

🎯 装饰器核心概念示意图
原函数 function() 装饰器 @decorator + 新增功能 增强函数 原功能 + 新功能 装饰器包装原函数,在不修改原代码的情况下添加新功能 遵循开放-封闭原则 (OCP)
🔧

代码复用

将通用功能抽象为装饰器,避免代码重复,提高可维护性

🎨

关注点分离

将横切关注点(如日志、权限)与业务逻辑解耦

动态增强

在运行时动态修改函数行为,无需改动源代码

📦

可组合性

多个装饰器可以叠加使用,灵活组合功能

02

工作原理

🔍 装饰器的本质

要理解装饰器,首先需要理解 Python 中函数是一等公民这一概念。函数可以:

  • 作为参数传递给其他函数
  • 作为其他函数的返回值
  • 赋值给变量
  • 存储在数据结构中
⚙️ 装饰器执行流程图
① 定义装饰器函数 ② 使用 @decorator 语法 ③ Python 自动执行转换 func = decorator(func) ④ 返回包装后的函数 关键步骤 调用被装饰的函数时,实际执行的是包装函数 wrapper wrapper 内部会调用原函数并添加额外逻辑

💡 核心理解

当你写 @decorator 放在函数定义上方时,Python 会自动将该函数作为参数传递给装饰器,并用装饰器的返回值替换原函数。这就是为什么装饰器必须返回一个可调用对象。

03

代码示例

🔰 示例1:基础装饰器 - 函数计时器

最基础的装饰器用法,用于测量函数执行时间:

Python
import time
from functools import wraps

# 定义一个计时装饰器
def timer(func):
    """测量函数执行时间的装饰器"""
    @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

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

# 调用函数
result = slow_function()
print(f"返回值: {result}")
📤 输出结果:
函数 slow_function 执行耗时: 1.0012 秒
返回值: 完成!

⚡ 示例2:带参数的装饰器 - 重试机制

当装饰器本身需要接收参数时,需要再嵌套一层函数:

Python
import time
from functools import wraps

# 带参数的重试装饰器
def retry(max_attempts=3, delay=1):
    """
    自动重试装饰器
    :param max_attempts: 最大重试次数
    :param delay: 重试间隔(秒)
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    print(f"第 {attempts} 次尝试失败: {e}")
                    if attempts < max_attempts:
                        print(f"等待 {delay} 秒后重试...")
                        time.sleep(delay)
            raise Exception(f"函数 {func.__name__} 在 {max_attempts} 次尝试后仍然失败")
        return wrapper
    return decorator

# 模拟不稳定的网络请求
attempt_count = 0

@retry(max_attempts=3, delay=0.5)
def unstable_network_call():
    """模拟不稳定的网络请求"""
    global attempt_count
    attempt_count += 1
    if attempt_count < 3:
        raise ConnectionError("网络连接失败")
    return "请求成功!"

# 测试重试机制
result = unstable_network_call()
print(f"最终结果: {result}")
📤 输出结果:
第 1 次尝试失败: 网络连接失败
等待 0.5 秒后重试...
第 2 次尝试失败: 网络连接失败
等待 0.5 秒后重试...
最终结果: 请求成功!

🚀 示例3:类装饰器 - 缓存机制

使用类实现装饰器,可以维护状态,实现更复杂的功能:

Python
from functools import wraps

class Memoize:
    """
    缓存装饰器类
    缓存函数的计算结果,避免重复计算
    """
    def __init__(self, func):
        self.func = func
        self.cache = {}  # 存储缓存结果
        wraps(func)(self)  # 保留原函数元信息
    
    def __call__(self, *args):
        # 检查缓存中是否已有结果
        if args in self.cache:
            print(f"  → 缓存命中: {args}")
            return self.cache[args]
        
        # 计算结果并存入缓存
        print(f"  → 计算中: {args}")
        result = self.func(*args)
        self.cache[args] = result
        return result
    
    def clear_cache(self):
        """清空缓存"""
        self.cache.clear()

# 使用缓存装饰器优化递归计算
@Memoize
def fibonacci(n):
    """计算斐波那契数列第 n 项"""
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# 测试缓存效果
print("计算 fibonacci(5):")
result = fibonacci(5)
print(f"结果: {result}\n")

print("再次计算 fibonacci(5):")
result = fibonacci(5)
print(f"结果: {result}")
📤 输出结果:
计算 fibonacci(5):
→ 计算中: (5,)
→ 计算中: (4,)
→ 计算中: (3,)
→ 计算中: (2,)
→ 计算中: (1,)
→ 计算中: (0,)
→ 缓存命中: (1,)
→ 缓存命中: (2,)
→ 缓存命中: (3,)
结果: 5

再次计算 fibonacci(5):
→ 缓存命中: (5,)
结果: 5
04

高级应用

🔗 多装饰器嵌套执行顺序
@decorator_a (外层) @decorator_b (中层) @decorator_c (内层) 原函数 func 执行顺序 ↓ 1. decorator_a 2. decorator_b 3. decorator_c 4. func() 装饰器从下往上应用,从外往内执行

🎯 常见应用场景

📝

日志记录

自动记录函数调用参数、返回值和执行时间

🔐

权限验证

在执行函数前检查用户权限,实现访问控制

💾

结果缓存

缓存函数计算结果,避免重复计算提升性能

参数验证

在函数执行前自动验证参数类型和范围

🔄

事务管理

数据库事务的自动提交和回滚处理

⏱️

限流控制

限制函数的调用频率,防止资源过度使用

05

最佳实践

✅ 使用要点

1

使用 @wraps

始终使用 functools.wraps 保留原函数的元信息(__name__, __doc__ 等)

2

支持所有参数

使用 *args 和 **kwargs 确保装饰器能处理任意参数的函数

3

保持简单

每个装饰器只做一件事,遵循单一职责原则

4

添加文档

为装饰器编写清晰的文档字符串说明其功能

⚠️ 常见陷阱

  • 忘记调用原函数或返回其结果
  • 带参数装饰器忘记额外的嵌套层
  • 装饰器副作用影响了函数的纯净性
  • 过度使用装饰器导致调试困难

💡 Python 标准库中的装饰器

@staticmethod - 定义静态方法
@classmethod - 定义类方法
@property - 将方法转为属性访问
@functools.lru_cache - 内置的缓存装饰器
@dataclasses.dataclass - 自动生成类方法

06

性能与调试

📊 装饰器性能影响

装饰器会增加额外的函数调用层次,需要注意性能开销:

Python
import time
from functools import wraps

# 性能测试装饰器
def benchmark(iterations=10000):
    """测试函数执行性能"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            for _ in range(iterations):
                result = func(*args, **kwargs)
            end = time.perf_counter()
            avg_time = (end - start) / iterations * 1000  # 毫秒
            print(f"{func.__name__}: {avg_time:.6f}ms 平均耗时 ({iterations}次调用)")
            return result
        return wrapper
    return decorator

# 比较使用装饰器和不使用装饰器的性能
def simple_add(a, b):
    return a + b

def heavy_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 模拟复杂的装饰器逻辑
        _ = [i for i in range(100)]  # 额外开销
        return func(*args, **kwargs)
    return wrapper

@heavy_decorator
def decorated_add(a, b):
    return a + b

# 性能比较
print("性能比较:")
benchmark()(lambda: simple_add(1, 2))()
benchmark()(lambda: decorated_add(1, 2))()

⚡ 性能优化建议

1. 避免过度嵌套:多层装饰器会增加调用栈深度
2. 使用内置装饰器:如 @lru_cache 经过高度优化
3. 懒惰计算:只在必要时执行装饰器逻辑
4. 缓存结果:对于相同输入,缓存计算结果

🐞 装饰器调试技巧

装饰器可能使调试变得复杂,以下是一些实用的调试方法:

Python
from functools import wraps
import inspect

# 调试装饰器:记录函数调用信息
def debug(func):
    """打印函数调用详细信息的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取调用位置
        frame = inspect.currentframe().f_back
        filename = frame.f_code.co_filename
        lineno = frame.f_lineno
        
        # 打印调用信息
        args_repr = [repr(a) for a in args]
        kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
        signature = ", ".join(args_repr + kwargs_repr)
        
        print(f"[DEBUG] 调用 {func.__name__}({signature})")
        print(f"        位置: {filename}:{lineno}")
        
        # 执行函数
        try:
            result = func(*args, **kwargs)
            print(f"[DEBUG] {func.__name__} 返回: {result!r}")
            return result
        except Exception as e:
            print(f"[DEBUG] {func.__name__} 抛出异常: {e!r}")
            raise
    
    return wrapper

# 使用示例
@debug
def calculate(x, y, operation="add"):
    if operation == "add":
        return x + y
    elif operation == "multiply":
        return x * y
    else:
        raise ValueError(f"未知操作: {operation}")

# 测试调试功能
result1 = calculate(10, 5)
result2 = calculate(10, 5, operation="multiply")
📤 输出结果:
[DEBUG] 调用 calculate(10, 5)
位置: example.py:25
[DEBUG] calculate 返回: 15
[DEBUG] 调用 calculate(10, 5, operation='multiply')
位置: example.py:26
[DEBUG] calculate 返回: 50

🔍 装饰器链追踪

查看函数被哪些装饰器包裹:

Python
from functools import wraps

def trace_decorators(func):
    """显示装饰器链的装饰器"""
    if not hasattr(func, '__decorator_chain__'):
        func.__decorator_chain__ = []
    
    def decorator(inner_func):
        @wraps(inner_func)
        def wrapper(*args, **kwargs):
            return inner_func(*args, **kwargs)
        
        # 记录装饰器
        if not hasattr(wrapper, '__decorator_chain__'):
            wrapper.__decorator_chain__ = getattr(inner_func, '__decorator_chain__', [])
        wrapper.__decorator_chain__.append(func.__name__)
        
        return wrapper
    
    return decorator

# 创建多个装饰器
@trace_decorators
def logger(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"📝 记录: {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@trace_decorators
def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"⏱️ 计时: {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

# 使用多个装饰器
@logger
@timer
def my_function():
    return "完成"

# 查看装饰器链
print(f"装饰器链: {my_function.__decorator_chain__}")
result = my_function()
📤 输出结果:
装饰器链: ['timer', 'logger']
📝 记录: my_function
⏱️ 计时: my_function

⚠️ 常见调试问题

  • 函数名丢失:忘记使用 @wraps 导致元信息丢失
  • 堆栈追踪混乱:多层装饰器使错误信息难以定位
  • 循环引用:装饰器互相嵌套导致死循环
  • 全局状态:装饰器使用全局变量导致难以测试
07

实战应用

🔒 示例1:API 访问频率限制

在 Web 应用中,需要限制用户对 API 的访问频率,防止滥用:

Python
import time
from functools import wraps
from collections import defaultdict

class RateLimiter:
    """
    访问频率限制装饰器
    在指定时间窗口内限制函数调用次数
    """
    def __init__(self, max_calls, time_window):
        """
        :param max_calls: 最大调用次数
        :param time_window: 时间窗口(秒)
        """
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = defaultdict(list)  # 存储每个用户的调用记录
    
    def __call__(self, func):
        @wraps(func)
        def wrapper(user_id, *args, **kwargs):
            now = time.time()
            
            # 清除超出时间窗口的记录
            self.calls[user_id] = [
                call_time for call_time in self.calls[user_id]
                if now - call_time < self.time_window
            ]
            
            # 检查是否超过频率限制
            if len(self.calls[user_id]) >= self.max_calls:
                remaining_time = int(self.time_window - (now - self.calls[user_id][0]))
                raise Exception(
                    f"访问频率限制:用户 {user_id} 请在 {remaining_time} 秒后重试"
                )
            
            # 记录本次调用
            self.calls[user_id].append(now)
            return func(user_id, *args, **kwargs)
        
        return wrapper

# 使用示例:限制 10 秒内最多调用 3 次
@RateLimiter(max_calls=3, time_window=10)
def api_endpoint(user_id, data):
    """模拟 API 接口"""
    return f"用户 {user_id} 请求成功:{data}"

# 测试限流功能
try:
    print(api_endpoint("user_001", "请求1"))
    print(api_endpoint("user_001", "请求2"))
    print(api_endpoint("user_001", "请求3"))
    print(api_endpoint("user_001", "请求4"))  # 这个会被拒绝
except Exception as e:
    print(f"错误:{e}")
📤 输出结果:
用户 user_001 请求成功:请求1
用户 user_001 请求成功:请求2
用户 user_001 请求成功:请求3
错误:访问频率限制:用户 user_001 请在 10 秒后重试

✅ 示例2:参数类型验证装饰器

自动验证函数参数的类型和值,减少样板代码:

Python
from functools import wraps
from typing import get_type_hints

def validate_types(func):
    """
    根据类型注解自动验证函数参数类型
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取函数的类型注解
        hints = get_type_hints(func)
        
        # 验证位置参数
        func_args = func.__code__.co_varnames[:func.__code__.co_argcount]
        for arg_name, arg_value in zip(func_args, args):
            if arg_name in hints:
                expected_type = hints[arg_name]
                if not isinstance(arg_value, expected_type):
                    raise TypeError(
                        f"参数 '{arg_name}' 类型错误:期望 {expected_type.__name__},实际得到 {type(arg_value).__name__}"
                    )
        
        # 验证关键字参数
        for arg_name, arg_value in kwargs.items():
            if arg_name in hints:
                expected_type = hints[arg_name]
                if not isinstance(arg_value, expected_type):
                    raise TypeError(
                        f"参数 '{arg_name}' 类型错误:期望 {expected_type.__name__},实际得到 {type(arg_value).__name__}"
                    )
        
        return func(*args, **kwargs)
    
    return wrapper

# 使用示例
@validate_types
def calculate_total(price: float, quantity: int, discount: float = 0.0) -> float:
    """计算总价(含折扣)"""
    return price * quantity * (1 - discount)

# 正确调用
print(calculate_total(99.9, 2, 0.1))  # 179.82

# 错误调用(类型不匹配)
try:
    print(calculate_total("99.9", 2))  # 将抛出 TypeError
except TypeError as e:
    print(f"错误:{e}")
📤 输出结果:
179.82
错误:参数 'price' 类型错误:期望 float,实际得到 str

💾 示例3:数据库事务管理

自动处理数据库事务的提交和回滚:

Python
from functools import wraps

def transactional(func):
    """
    数据库事务装饰器
    自动处理事务的提交和回滚
    """
    @wraps(func)
    def wrapper(db_connection, *args, **kwargs):
        try:
            # 开始事务
            print("[事务] 开始")
            
            # 执行业务逻辑
            result = func(db_connection, *args, **kwargs)
            
            # 提交事务
            db_connection.commit()
            print("[事务] 提交成功")
            
            return result
        
        except Exception as e:
            # 发生错误,回滚事务
            db_connection.rollback()
            print(f"[事务] 回滚:{e}")
            raise
    
    return wrapper

# 模拟数据库连接
class MockDBConnection:
    def commit(self):
        print("  -> DB: 执行 COMMIT")
    
    def rollback(self):
        print("  -> DB: 执行 ROLLBACK")
    
    def execute(self, sql):
        print(f"  -> DB: 执行 SQL: {sql}")

# 使用事务装饰器
@transactional
def transfer_money(db, from_account, to_account, amount):
    """转账操作"""
    db.execute(f"UPDATE accounts SET balance = balance - {amount} WHERE id = {from_account}")
    db.execute(f"UPDATE accounts SET balance = balance + {amount} WHERE id = {to_account}")
    return f"从账户 {from_account} 向 {to_account} 转账 {amount} 元"

# 测试事务功能
db = MockDBConnection()
result = transfer_money(db, 1001, 1002, 100)
print(f"结果:{result}")
📤 输出结果:
[事务] 开始
-> DB: 执行 SQL: UPDATE accounts SET balance = balance - 100 WHERE id = 1001
-> DB: 执行 SQL: UPDATE accounts SET balance = balance + 100 WHERE id = 1002
-> DB: 执行 COMMIT
[事务] 提交成功
结果:从账户 1001 向 1002 转账 100 元
08

标准库装饰器

📦 Python 内置装饰器详解

Python 标准库提供了多个常用装饰器,了解它们可以帮助你写出更优雅的代码:

🔑 @property - 属性装饰器

将方法转换为只读属性,实现封装和数据验证:

Python
class Circle:
    def __init__(self, radius):
        self._radius = radius  # 私有属性
    
    @property
    def radius(self):
        """获取半径"""
        return self._radius
    
    @radius.setter
    def radius(self, value):
        """设置半径,带数据验证"""
        if value <= 0:
            raise ValueError("半径必须大于 0")
        self._radius = value
    
    @property
    def area(self):
        """计算面积(只读属性)"""
        import math
        return math.pi * self._radius ** 2
    
    @property
    def circumference(self):
        """计算周长(只读属性)"""
        import math
        return 2 * math.pi * self._radius

# 使用示例
c = Circle(5)
print(f"半径:{c.radius}")  # 像访问属性一样
print(f"面积:{c.area:.2f}")
print(f"周长:{c.circumference:.2f}")

c.radius = 10  # 通过 setter 修改
print(f"新面积:{c.area:.2f}")
📤 输出结果:
半径:5
面积:78.54
周长:31.42
新面积:314.16

🏛️ @staticmethod 和 @classmethod

用于定义类级别的方法,不需要实例化:

Python
class DateUtil:
    """日期工具类"""
    
    # 类变量
    date_format = "%Y-%m-%d"
    
    @staticmethod
    def is_leap_year(year):
        """
        静态方法:判断是否为闰年
        不需要访问类或实例属性
        """
        return (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0)
    
    @classmethod
    def from_string(cls, date_string):
        """
        类方法:从字符串创建日期对象
        可以访问类属性
        """
        from datetime import datetime
        date_obj = datetime.strptime(date_string, cls.date_format)
        return cls(date_obj.year, date_obj.month, date_obj.day)
    
    @classmethod
    def today(cls):
        """类方法:获取今天的日期"""
        from datetime import datetime
        now = datetime.now()
        return cls(now.year, now.month, now.day)
    
    def __init__(self, year, month, day):
        self.year = year
        self.month = month
        self.day = day
    
    def __str__(self):
        return f"{self.year}-{self.month:02d}-{self.day:02d}"

# 使用静态方法(不需要实例)
print(DateUtil.is_leap_year(2024))  # True
print(DateUtil.is_leap_year(2023))  # False

# 使用类方法创建对象
date1 = DateUtil.from_string("2024-12-25")
print(f"解析的日期:{date1}")

date2 = DateUtil.today()
print(f"今天是:{date2}")

💡 关键区别

@staticmethod:不需要访问类或实例属性,类似于普通函数
@classmethod:第一个参数是类本身(cls),可以访问类属性和创建实例

⚡ @functools.lru_cache - 内置缓存装饰器

Python 标准库提供的高效缓存装饰器,使用 LRU(Least Recently Used)算法:

Python
from functools import lru_cache
import time

# 使用 LRU 缓存,最多缓存 128 个结果
@lru_cache(maxsize=128)
def expensive_computation(n):
    """模拟耗时计算"""
    print(f"  计算 {n}...")
    time.sleep(1)  # 模拟耗时操作
    return n * n

print("第一次调用:")
start = time.time()
result1 = expensive_computation(5)
print(f"结果:{result1},耗时:{time.time() - start:.2f}秒\n")

print("第二次调用(缓存命中):")
start = time.time()
result2 = expensive_computation(5)
print(f"结果:{result2},耗时:{time.time() - start:.4f}秒\n")

# 查看缓存统计
print(f"缓存信息:{expensive_computation.cache_info()}")

# 清空缓存
expensive_computation.cache_clear()
print("缓存已清空")
📤 输出结果:
第一次调用:
计算 5...
结果:25,耗时:1.00秒

第二次调用(缓存命中):
结果:25,耗时:0.0001秒

缓存信息:CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
缓存已清空

📝 总结

标准库装饰器提供了丰富的功能:
@property - 属性封装
@staticmethod / @classmethod - 类方法
@functools.lru_cache - 结果缓存
@functools.wraps - 保留元信息
@dataclasses.dataclass - 数据类
@contextlib.contextmanager - 上下文管理器